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

Add ArgumentAction.ACTION, support POSIX bundling in CAP, Move all Actions to their own file

This commit is contained in:
Roland Thomas Jr 2025-05-25 19:25:32 -04:00
parent 429b434566
commit fb1ffbe9f6
Signed by: roland
GPG Key ID: 7C3C2B085A4C2872
46 changed files with 1630 additions and 842 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
examples/action_example.py
View File

@ -1,6 +1,6 @@
import asyncio
from falyx import Action, ActionGroup, ChainedAction
from falyx.action import Action, ActionGroup, ChainedAction
# Actions can be defined as synchronous functions

11
examples/auto_args_group.py
View File

@ -1,6 +1,7 @@
import asyncio
from falyx import Action, ActionGroup, Command, Falyx
from falyx import Falyx
from falyx.action import Action, ActionGroup
# Define a shared async function
@ -19,10 +20,11 @@ action3 = Action("say_hello_3", action=say_hello)
# Combine into an ActionGroup
group = ActionGroup(name="greet_group", actions=[action1, action2, action3])
# Create the Command with auto_args=True
cmd = Command(
flx = Falyx("Test Group")
flx.add_command(
key="G",
description="Greet someone with multiple variations.",
aliases=["greet", "hello"],
action=group,
arg_metadata={
"name": {
@ -33,7 +35,4 @@ cmd = Command(
},
},
)
flx = Falyx("Test Group")
flx.add_command_from_command(cmd)
asyncio.run(flx.run())

3
examples/auto_parse_demo.py
View File

@ -1,6 +1,7 @@
import asyncio
from falyx import Action, ChainedAction, Falyx
from falyx import Falyx
from falyx.action import Action, ChainedAction
from falyx.utils import setup_logging
setup_logging()

4
examples/http_demo.py
View File

@ -2,8 +2,8 @@ import asyncio
from rich.console import Console
from falyx import ActionGroup, Falyx
from falyx.action import HTTPAction
from falyx import Falyx
from falyx.action import ActionGroup, HTTPAction
from falyx.hooks import ResultReporter
console = Console()

3
examples/pipeline_demo.py
View File

@ -1,8 +1,7 @@
import asyncio
from falyx import Action, ActionGroup, ChainedAction
from falyx import ExecutionRegistry as er
from falyx import ProcessAction
from falyx.action import Action, ActionGroup, ChainedAction, ProcessAction
from falyx.retry import RetryHandler, RetryPolicy

3
examples/process_pool.py
View File

@ -1,6 +1,7 @@
from rich.console import Console
from falyx import Falyx, ProcessAction
from falyx import Falyx
from falyx.action import ProcessAction
from falyx.themes import NordColors as nc
console = Console()

3
examples/run_key.py
View File

@ -1,6 +1,7 @@
import asyncio
from falyx import Action, Falyx
from falyx import Falyx
from falyx.action import Action
async def main():

4
examples/shell_example.py
View File

@ -1,8 +1,8 @@
#!/usr/bin/env python
import asyncio
from falyx import Action, ChainedAction, Falyx
from falyx.action import ShellAction
from falyx import Falyx
from falyx.action import Action, ChainedAction, ShellAction
from falyx.hooks import ResultReporter
from falyx.utils import setup_logging

3
examples/simple.py
View File

@ -1,7 +1,8 @@
import asyncio
import random
from falyx import Action, ChainedAction, Falyx
from falyx import Falyx
from falyx.action import Action, ChainedAction
from falyx.utils import setup_logging
setup_logging()

3
examples/submenu.py
View File

@ -1,7 +1,8 @@
import asyncio
import random
from falyx import Action, ChainedAction, Falyx
from falyx import Falyx
from falyx.action import Action, ChainedAction
from falyx.utils import setup_logging
setup_logging()

11
falyx/__init__.py
View File

@ -7,23 +7,12 @@ Licensed under the MIT License. See LICENSE file for details.
import logging
from .action.action import Action, ActionGroup, ChainedAction, ProcessAction
from .command import Command
from .context import ExecutionContext, SharedContext
from .execution_registry import ExecutionRegistry
from .falyx import Falyx
logger = logging.getLogger("falyx")
__all__ = [
"Action",
"ChainedAction",
"ActionGroup",
"ProcessAction",
"Falyx",
"Command",
"ExecutionContext",
"SharedContext",
"ExecutionRegistry",
"HookType",
]

16
falyx/action/__init__.py
View File

@ -5,19 +5,17 @@ Copyright (c) 2025 rtj.dev LLC.
Licensed under the MIT License. See LICENSE file for details.
"""
from .action import (
Action,
ActionGroup,
BaseAction,
ChainedAction,
FallbackAction,
LiteralInputAction,
ProcessAction,
)
from .action import Action
from .action_factory import ActionFactoryAction
from .action_group import ActionGroup
from .base import BaseAction
from .chained_action import ChainedAction
from .fallback_action import FallbackAction
from .http_action import HTTPAction
from .io_action import BaseIOAction, ShellAction
from .literal_input_action import LiteralInputAction
from .menu_action import MenuAction
from .process_action import ProcessAction
from .prompt_menu_action import PromptMenuAction
from .select_file_action import SelectFileAction
from .selection_action import SelectionAction

728
falyx/action/action.py
View File

@ -1,170 +1,21 @@
# Falyx CLI Framework — (c) 2025 rtj.dev LLC — MIT Licensed
"""action.py
Core action system for Falyx.
This module defines the building blocks for executable actions and workflows,
providing a structured way to compose, execute, recover, and manage sequences of
operations.
All actions are callable and follow a unified signature:
result = action(*args, **kwargs)
Core guarantees:
- Full hook lifecycle support (before, on_success, on_error, after, on_teardown).
- Consistent timing and execution context tracking for each run.
- Unified, predictable result handling and error propagation.
- Optional last_result injection to enable flexible, data-driven workflows.
- Built-in support for retries, rollbacks, parallel groups, chaining, and fallback
recovery.
Key components:
- Action: wraps a function or coroutine into a standard executable unit.
- ChainedAction: runs actions sequentially, optionally injecting last results.
- ActionGroup: runs actions in parallel and gathers results.
- ProcessAction: executes CPU-bound functions in a separate process.
- LiteralInputAction: injects static values into workflows.
- FallbackAction: gracefully recovers from failures or missing data.
This design promotes clean, fault-tolerant, modular CLI and automation systems.
"""
"""action.py"""
from __future__ import annotations
import asyncio
import random
from abc import ABC, abstractmethod
from concurrent.futures import ProcessPoolExecutor
from functools import cached_property, partial
from typing import Any, Callable
from rich.console import Console
from rich.tree import Tree
from falyx.context import ExecutionContext, SharedContext
from falyx.debug import register_debug_hooks
from falyx.exceptions import EmptyChainError
from falyx.action.base import BaseAction
from falyx.context import ExecutionContext
from falyx.execution_registry import ExecutionRegistry as er
from falyx.hook_manager import Hook, HookManager, HookType
from falyx.hook_manager import 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
class BaseAction(ABC):
"""
Base class for actions. Actions can be simple functions or more
complex actions like `ChainedAction` or `ActionGroup`. They can also
be run independently or as part of Falyx.
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').
"""
def __init__(
self,
name: str,
*,
hooks: HookManager | None = None,
inject_last_result: bool = False,
inject_into: str = "last_result",
never_prompt: bool = False,
logging_hooks: bool = False,
) -> None:
self.name = name
self.hooks = hooks or HookManager()
self.is_retryable: bool = False
self.shared_context: SharedContext | None = None
self.inject_last_result: bool = inject_last_result
self.inject_into: str = inject_into
self._never_prompt: bool = never_prompt
self._skip_in_chain: bool = False
self.console = Console(color_system="auto")
self.options_manager: OptionsManager | None = None
if logging_hooks:
register_debug_hooks(self.hooks)
async def __call__(self, *args, **kwargs) -> Any:
return await self._run(*args, **kwargs)
@abstractmethod
async def _run(self, *args, **kwargs) -> Any:
raise NotImplementedError("_run must be implemented by subclasses")
@abstractmethod
async def preview(self, parent: Tree | None = None):
raise NotImplementedError("preview must be implemented by subclasses")
@abstractmethod
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.
"""
raise NotImplementedError("get_infer_target must be implemented by subclasses")
def set_options_manager(self, options_manager: OptionsManager) -> None:
self.options_manager = options_manager
def set_shared_context(self, shared_context: SharedContext) -> None:
self.shared_context = shared_context
def get_option(self, option_name: str, default: Any = None) -> Any:
"""
Resolve an option from the OptionsManager if present, otherwise use the fallback.
"""
if self.options_manager:
return self.options_manager.get(option_name, default)
return default
@property
def last_result(self) -> Any:
"""Return the last result from the shared context."""
if self.shared_context:
return self.shared_context.last_result()
return None
@property
def never_prompt(self) -> bool:
return self.get_option("never_prompt", self._never_prompt)
def prepare(
self, shared_context: SharedContext, options_manager: OptionsManager | None = None
) -> BaseAction:
"""
Prepare the action specifically for sequential (ChainedAction) execution.
Can be overridden for chain-specific logic.
"""
self.set_shared_context(shared_context)
if options_manager:
self.set_options_manager(options_manager)
return self
def _maybe_inject_last_result(self, kwargs: dict[str, Any]) -> dict[str, Any]:
if self.inject_last_result and self.shared_context:
key = self.inject_into
if key in kwargs:
logger.warning("[%s] Overriding '%s' with last_result", self.name, key)
kwargs = dict(kwargs)
kwargs[key] = self.shared_context.last_result()
return kwargs
def register_hooks_recursively(self, hook_type: HookType, hook: Hook):
"""Register a hook for all actions and sub-actions."""
self.hooks.register(hook_type, hook)
async def _write_stdout(self, data: str) -> None:
"""Override in subclasses that produce terminal output."""
def __repr__(self) -> str:
return str(self)
class Action(BaseAction):
"""
Action wraps a simple function or coroutine into a standard executable unit.
@ -309,574 +160,3 @@ class Action(BaseAction):
f"args={self.args!r}, kwargs={self.kwargs!r}, "
f"retry={self.retry_policy.enabled})"
)
class LiteralInputAction(Action):
"""
LiteralInputAction injects a static value into a ChainedAction.
This allows embedding hardcoded values mid-pipeline, useful when:
- Providing default or fallback inputs.
- Starting a pipeline with a fixed input.
- Supplying missing context manually.
Args:
value (Any): The static value to inject.
"""
def __init__(self, value: Any):
self._value = value
async def literal(*_, **__):
return value
super().__init__("Input", literal)
@cached_property
def value(self) -> Any:
"""Return the literal value."""
return self._value
async def preview(self, parent: Tree | None = None):
label = [f"[{OneColors.LIGHT_YELLOW}]📥 LiteralInput[/] '{self.name}'"]
label.append(f" [dim](value = {repr(self.value)})[/dim]")
if parent:
parent.add("".join(label))
else:
self.console.print(Tree("".join(label)))
def __str__(self) -> str:
return f"LiteralInputAction(value={self.value!r})"
class FallbackAction(Action):
"""
FallbackAction provides a default value if the previous action failed or
returned None.
It injects the last result and checks:
- If last_result is not None, it passes it through unchanged.
- If last_result is None (e.g., due to failure), it replaces it with a fallback value.
Used in ChainedAction pipelines to gracefully recover from errors or missing data.
When activated, it consumes the preceding error and allows the chain to continue
normally.
Args:
fallback (Any): The fallback value to use if last_result is None.
"""
def __init__(self, fallback: Any):
self._fallback = fallback
async def _fallback_logic(last_result):
return last_result if last_result is not None else fallback
super().__init__(name="Fallback", action=_fallback_logic, inject_last_result=True)
@cached_property
def fallback(self) -> Any:
"""Return the fallback value."""
return self._fallback
async def preview(self, parent: Tree | None = None):
label = [f"[{OneColors.LIGHT_RED}]🛟 Fallback[/] '{self.name}'"]
label.append(f" [dim](uses fallback = {repr(self.fallback)})[/dim]")
if parent:
parent.add("".join(label))
else:
self.console.print(Tree("".join(label)))
def __str__(self) -> str:
return f"FallbackAction(fallback={self.fallback!r})"
class ActionListMixin:
"""Mixin for managing a list of actions."""
def __init__(self) -> None:
self.actions: list[BaseAction] = []
def set_actions(self, actions: list[BaseAction]) -> None:
"""Replaces the current action list with a new one."""
self.actions.clear()
for action in actions:
self.add_action(action)
def add_action(self, action: BaseAction) -> None:
"""Adds an action to the list."""
self.actions.append(action)
def remove_action(self, name: str) -> None:
"""Removes an action by name."""
self.actions = [action for action in self.actions if action.name != name]
def has_action(self, name: str) -> bool:
"""Checks if an action with the given name exists."""
return any(action.name == name for action in self.actions)
def get_action(self, name: str) -> BaseAction | None:
"""Retrieves an action by name."""
for action in self.actions:
if action.name == name:
return action
return None
class ChainedAction(BaseAction, ActionListMixin):
"""
ChainedAction executes a sequence of actions one after another.
Features:
- Supports optional automatic last_result injection (auto_inject).
- Recovers from intermediate errors using FallbackAction if present.
- Rolls back all previously executed actions if a failure occurs.
- Handles literal values with LiteralInputAction.
Best used for defining robust, ordered workflows where each step can depend on
previous results.
Args:
name (str): Name of the chain.
actions (list): List of actions or literals to execute.
hooks (HookManager, optional): Hooks for lifecycle events.
inject_last_result (bool, optional): Whether to inject last results into kwargs
by default.
inject_into (str, optional): Key name for injection.
auto_inject (bool, optional): Auto-enable injection for subsequent actions.
return_list (bool, optional): Whether to return a list of all results. False
returns the last result.
"""
def __init__(
self,
name: str,
actions: list[BaseAction | Any] | None = None,
*,
hooks: HookManager | None = None,
inject_last_result: bool = False,
inject_into: str = "last_result",
auto_inject: bool = False,
return_list: bool = False,
) -> None:
super().__init__(
name,
hooks=hooks,
inject_last_result=inject_last_result,
inject_into=inject_into,
)
ActionListMixin.__init__(self)
self.auto_inject = auto_inject
self.return_list = return_list
if actions:
self.set_actions(actions)
def _wrap_if_needed(self, action: BaseAction | Any) -> BaseAction:
if isinstance(action, BaseAction):
return action
elif callable(action):
return Action(name=action.__name__, action=action)
else:
return LiteralInputAction(action)
def add_action(self, action: BaseAction | Any) -> None:
action = self._wrap_if_needed(action)
if self.actions and self.auto_inject and not action.inject_last_result:
action.inject_last_result = True
super().add_action(action)
if hasattr(action, "register_teardown") and callable(action.register_teardown):
action.register_teardown(self.hooks)
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, 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.")
shared_context = SharedContext(name=self.name, action=self)
if self.shared_context:
shared_context.add_result(self.shared_context.last_result())
updated_kwargs = self._maybe_inject_last_result(kwargs)
context = ExecutionContext(
name=self.name,
args=args,
kwargs=updated_kwargs,
action=self,
extra={"results": [], "rollback_stack": []},
shared_context=shared_context,
)
context.start_timer()
try:
await self.hooks.trigger(HookType.BEFORE, context)
for index, action in enumerate(self.actions):
if action._skip_in_chain:
logger.debug(
"[%s] Skipping consumed action '%s'", self.name, action.name
)
continue
shared_context.current_index = index
prepared = action.prepare(shared_context, self.options_manager)
try:
result = await prepared(*args, **updated_kwargs)
except Exception as error:
if index + 1 < len(self.actions) and isinstance(
self.actions[index + 1], FallbackAction
):
logger.warning(
"[%s] Fallback triggered: %s, recovering with fallback "
"'%s'.",
self.name,
error,
self.actions[index + 1].name,
)
shared_context.add_result(None)
context.extra["results"].append(None)
fallback = self.actions[index + 1].prepare(shared_context)
result = await fallback()
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)
all_results = context.extra["results"]
assert (
all_results
), f"[{self.name}] No results captured. Something seriously went wrong."
context.result = all_results if self.return_list else all_results[-1]
await self.hooks.trigger(HookType.ON_SUCCESS, context)
return context.result
except Exception as error:
context.exception = error
shared_context.add_error(shared_context.current_index, error)
await self._rollback(context.extra["rollback_stack"], *args, **kwargs)
await self.hooks.trigger(HookType.ON_ERROR, context)
raise
finally:
context.stop_timer()
await self.hooks.trigger(HookType.AFTER, context)
await self.hooks.trigger(HookType.ON_TEARDOWN, context)
er.record(context)
async def _rollback(self, rollback_stack, *args, **kwargs):
"""
Roll back all executed actions in reverse order.
Rollbacks run even if a fallback recovered from failure,
ensuring consistent undo of all side effects.
Actions without rollback handlers are skipped.
Args:
rollback_stack (list): Actions to roll back.
*args, **kwargs: Passed to rollback handlers.
"""
for action in reversed(rollback_stack):
rollback = getattr(action, "rollback", None)
if rollback:
try:
logger.warning("[%s] Rolling back...", action.name)
await action.rollback(*args, **kwargs)
except Exception as error:
logger.error("[%s] Rollback failed: %s", action.name, error)
def register_hooks_recursively(self, hook_type: HookType, hook: Hook):
"""Register a hook for all actions and sub-actions."""
self.hooks.register(hook_type, hook)
for action in self.actions:
action.register_hooks_recursively(hook_type, hook)
async def preview(self, parent: Tree | None = None):
label = [f"[{OneColors.CYAN_b}]⛓ ChainedAction[/] '{self.name}'"]
if self.inject_last_result:
label.append(f" [dim](injects '{self.inject_into}')[/dim]")
tree = parent.add("".join(label)) if parent else Tree("".join(label))
for action in self.actions:
await action.preview(parent=tree)
if not parent:
self.console.print(tree)
def __str__(self):
return (
f"ChainedAction(name={self.name!r}, "
f"actions={[a.name for a in self.actions]!r}, "
f"auto_inject={self.auto_inject}, return_list={self.return_list})"
)
class ActionGroup(BaseAction, ActionListMixin):
"""
ActionGroup executes multiple actions concurrently in parallel.
It is ideal for independent tasks that can be safely run simultaneously,
improving overall throughput and responsiveness of workflows.
Core features:
- Parallel execution of all contained actions.
- Shared last_result injection across all actions if configured.
- Aggregated collection of individual results as (name, result) pairs.
- Hook lifecycle support (before, on_success, on_error, after, on_teardown).
- Error aggregation: captures all action errors and reports them together.
Behavior:
- If any action fails, the group collects the errors but continues executing
other actions without interruption.
- After all actions complete, ActionGroup raises a single exception summarizing
all failures, or returns all results if successful.
Best used for:
- Batch processing multiple independent tasks.
- Reducing latency for workflows with parallelizable steps.
- Isolating errors while maximizing successful execution.
Args:
name (str): Name of the chain.
actions (list): List of actions or literals to execute.
hooks (HookManager, optional): Hooks for lifecycle events.
inject_last_result (bool, optional): Whether to inject last results into kwargs
by default.
inject_into (str, optional): Key name for injection.
"""
def __init__(
self,
name: str,
actions: list[BaseAction] | None = None,
*,
hooks: HookManager | None = None,
inject_last_result: bool = False,
inject_into: str = "last_result",
):
super().__init__(
name,
hooks=hooks,
inject_last_result=inject_last_result,
inject_into=inject_into,
)
ActionListMixin.__init__(self)
if actions:
self.set_actions(actions)
def _wrap_if_needed(self, action: BaseAction | Any) -> BaseAction:
if isinstance(action, BaseAction):
return action
elif callable(action):
return Action(name=action.__name__, action=action)
else:
raise TypeError(
"ActionGroup only accepts BaseAction or callable, got "
f"{type(action).__name__}"
)
def add_action(self, action: BaseAction | Any) -> None:
action = self._wrap_if_needed(action)
super().add_action(action)
if hasattr(action, "register_teardown") and callable(action.register_teardown):
action.register_teardown(self.hooks)
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()
logger.debug(
"[%s] auto_args disabled: mismatched ActionGroup arguments",
self.name,
)
return None, 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:
shared_context.set_shared_result(self.shared_context.last_result())
updated_kwargs = self._maybe_inject_last_result(kwargs)
context = ExecutionContext(
name=self.name,
args=args,
kwargs=updated_kwargs,
action=self,
extra={"results": [], "errors": []},
shared_context=shared_context,
)
async def run_one(action: BaseAction):
try:
prepared = action.prepare(shared_context, self.options_manager)
result = await prepared(*args, **updated_kwargs)
shared_context.add_result((action.name, result))
context.extra["results"].append((action.name, result))
except Exception as error:
shared_context.add_error(shared_context.current_index, error)
context.extra["errors"].append((action.name, error))
context.start_timer()
try:
await self.hooks.trigger(HookType.BEFORE, context)
await asyncio.gather(*[run_one(a) for a in self.actions])
if context.extra["errors"]:
context.exception = Exception(
f"{len(context.extra['errors'])} action(s) failed: "
f"{' ,'.join(name for name, _ in context.extra['errors'])}"
)
await self.hooks.trigger(HookType.ON_ERROR, context)
raise context.exception
context.result = context.extra["results"]
await self.hooks.trigger(HookType.ON_SUCCESS, context)
return context.result
except Exception as error:
context.exception = error
raise
finally:
context.stop_timer()
await self.hooks.trigger(HookType.AFTER, context)
await self.hooks.trigger(HookType.ON_TEARDOWN, context)
er.record(context)
def register_hooks_recursively(self, hook_type: HookType, hook: Hook):
"""Register a hook for all actions and sub-actions."""
super().register_hooks_recursively(hook_type, hook)
for action in self.actions:
action.register_hooks_recursively(hook_type, hook)
async def preview(self, parent: Tree | None = None):
label = [f"[{OneColors.MAGENTA_b}]⏩ ActionGroup (parallel)[/] '{self.name}'"]
if self.inject_last_result:
label.append(f" [dim](receives '{self.inject_into}')[/dim]")
tree = parent.add("".join(label)) if parent else Tree("".join(label))
actions = self.actions.copy()
random.shuffle(actions)
await asyncio.gather(*(action.preview(parent=tree) for action in actions))
if not parent:
self.console.print(tree)
def __str__(self):
return (
f"ActionGroup(name={self.name!r}, actions={[a.name for a in self.actions]!r},"
f" inject_last_result={self.inject_last_result})"
)
class ProcessAction(BaseAction):
"""
ProcessAction runs a function in a separate process using ProcessPoolExecutor.
Features:
- Executes CPU-bound or blocking tasks without blocking the main event loop.
- Supports last_result injection into the subprocess.
- Validates that last_result is pickleable when injection is enabled.
Args:
name (str): Name of the action.
func (Callable): Function to execute in a new process.
args (tuple, optional): Positional arguments.
kwargs (dict, optional): Keyword arguments.
hooks (HookManager, optional): Hook manager for lifecycle events.
executor (ProcessPoolExecutor, optional): Custom executor if desired.
inject_last_result (bool, optional): Inject last result into the function.
inject_into (str, optional): Name of the injected key.
"""
def __init__(
self,
name: str,
action: Callable[..., Any],
*,
args: tuple = (),
kwargs: dict[str, Any] | None = None,
hooks: HookManager | None = None,
executor: ProcessPoolExecutor | None = None,
inject_last_result: bool = False,
inject_into: str = "last_result",
):
super().__init__(
name,
hooks=hooks,
inject_last_result=inject_last_result,
inject_into=inject_into,
)
self.action = action
self.args = args
self.kwargs = kwargs or {}
self.executor = executor or ProcessPoolExecutor()
self.is_retryable = True
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:
last_result = self.shared_context.last_result()
if not self._validate_pickleable(last_result):
raise ValueError(
f"Cannot inject last result into {self.name}: "
f"last result is not pickleable."
)
combined_args = args + self.args
combined_kwargs = self._maybe_inject_last_result({**self.kwargs, **kwargs})
context = ExecutionContext(
name=self.name,
args=combined_args,
kwargs=combined_kwargs,
action=self,
)
loop = asyncio.get_running_loop()
context.start_timer()
try:
await self.hooks.trigger(HookType.BEFORE, context)
result = await loop.run_in_executor(
self.executor, partial(self.action, *combined_args, **combined_kwargs)
)
context.result = result
await self.hooks.trigger(HookType.ON_SUCCESS, context)
return result
except Exception as error:
context.exception = error
await self.hooks.trigger(HookType.ON_ERROR, context)
if context.result is not None:
return context.result
raise
finally:
context.stop_timer()
await self.hooks.trigger(HookType.AFTER, context)
await self.hooks.trigger(HookType.ON_TEARDOWN, context)
er.record(context)
def _validate_pickleable(self, obj: Any) -> bool:
try:
import pickle
pickle.dumps(obj)
return True
except (pickle.PicklingError, TypeError):
return False
async def preview(self, parent: Tree | None = None):
label = [
f"[{OneColors.DARK_YELLOW_b}]🧠 ProcessAction (new process)[/] '{self.name}'"
]
if self.inject_last_result:
label.append(f" [dim](injects '{self.inject_into}')[/dim]")
if parent:
parent.add("".join(label))
else:
self.console.print(Tree("".join(label)))
def __str__(self) -> str:
return (
f"ProcessAction(name={self.name!r}, "
f"action={getattr(self.action, '__name__', repr(self.action))}, "
f"args={self.args!r}, kwargs={self.kwargs!r})"
)

2
falyx/action/action_factory.py
View File

@ -4,7 +4,7 @@ from typing import Any, Callable
from rich.tree import Tree
from falyx.action.action import BaseAction
from falyx.action.base import BaseAction
from falyx.context import ExecutionContext
from falyx.execution_registry import ExecutionRegistry as er
from falyx.hook_manager import HookType

169
falyx/action/action_group.py Normal file
View File

@ -0,0 +1,169 @@
import asyncio
import random
from typing import Any, Callable
from rich.tree import Tree
from falyx.action.action import Action
from falyx.action.base import BaseAction
from falyx.action.mixins import ActionListMixin
from falyx.context import ExecutionContext, SharedContext
from falyx.execution_registry import ExecutionRegistry as er
from falyx.hook_manager import Hook, HookManager, HookType
from falyx.logger import logger
from falyx.parsers.utils import same_argument_definitions
from falyx.themes.colors import OneColors
class ActionGroup(BaseAction, ActionListMixin):
"""
ActionGroup executes multiple actions concurrently in parallel.
It is ideal for independent tasks that can be safely run simultaneously,
improving overall throughput and responsiveness of workflows.
Core features:
- Parallel execution of all contained actions.
- Shared last_result injection across all actions if configured.
- Aggregated collection of individual results as (name, result) pairs.
- Hook lifecycle support (before, on_success, on_error, after, on_teardown).
- Error aggregation: captures all action errors and reports them together.
Behavior:
- If any action fails, the group collects the errors but continues executing
other actions without interruption.
- After all actions complete, ActionGroup raises a single exception summarizing
all failures, or returns all results if successful.
Best used for:
- Batch processing multiple independent tasks.
- Reducing latency for workflows with parallelizable steps.
- Isolating errors while maximizing successful execution.
Args:
name (str): Name of the chain.
actions (list): List of actions or literals to execute.
hooks (HookManager, optional): Hooks for lifecycle events.
inject_last_result (bool, optional): Whether to inject last results into kwargs
by default.
inject_into (str, optional): Key name for injection.
"""
def __init__(
self,
name: str,
actions: list[BaseAction] | None = None,
*,
hooks: HookManager | None = None,
inject_last_result: bool = False,
inject_into: str = "last_result",
):
super().__init__(
name,
hooks=hooks,
inject_last_result=inject_last_result,
inject_into=inject_into,
)
ActionListMixin.__init__(self)
if actions:
self.set_actions(actions)
def _wrap_if_needed(self, action: BaseAction | Any) -> BaseAction:
if isinstance(action, BaseAction):
return action
elif callable(action):
return Action(name=action.__name__, action=action)
else:
raise TypeError(
"ActionGroup only accepts BaseAction or callable, got "
f"{type(action).__name__}"
)
def add_action(self, action: BaseAction | Any) -> None:
action = self._wrap_if_needed(action)
super().add_action(action)
if hasattr(action, "register_teardown") and callable(action.register_teardown):
action.register_teardown(self.hooks)
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()
logger.debug(
"[%s] auto_args disabled: mismatched ActionGroup arguments",
self.name,
)
return None, 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:
shared_context.set_shared_result(self.shared_context.last_result())
updated_kwargs = self._maybe_inject_last_result(kwargs)
context = ExecutionContext(
name=self.name,
args=args,
kwargs=updated_kwargs,
action=self,
extra={"results": [], "errors": []},
shared_context=shared_context,
)
async def run_one(action: BaseAction):
try:
prepared = action.prepare(shared_context, self.options_manager)
result = await prepared(*args, **updated_kwargs)
shared_context.add_result((action.name, result))
context.extra["results"].append((action.name, result))
except Exception as error:
shared_context.add_error(shared_context.current_index, error)
context.extra["errors"].append((action.name, error))
context.start_timer()
try:
await self.hooks.trigger(HookType.BEFORE, context)
await asyncio.gather(*[run_one(a) for a in self.actions])
if context.extra["errors"]:
context.exception = Exception(
f"{len(context.extra['errors'])} action(s) failed: "
f"{' ,'.join(name for name, _ in context.extra['errors'])}"
)
await self.hooks.trigger(HookType.ON_ERROR, context)
raise context.exception
context.result = context.extra["results"]
await self.hooks.trigger(HookType.ON_SUCCESS, context)
return context.result
except Exception as error:
context.exception = error
raise
finally:
context.stop_timer()
await self.hooks.trigger(HookType.AFTER, context)
await self.hooks.trigger(HookType.ON_TEARDOWN, context)
er.record(context)
def register_hooks_recursively(self, hook_type: HookType, hook: Hook):
"""Register a hook for all actions and sub-actions."""
super().register_hooks_recursively(hook_type, hook)
for action in self.actions:
action.register_hooks_recursively(hook_type, hook)
async def preview(self, parent: Tree | None = None):
label = [f"[{OneColors.MAGENTA_b}]⏩ ActionGroup (parallel)[/] '{self.name}'"]
if self.inject_last_result:
label.append(f" [dim](receives '{self.inject_into}')[/dim]")
tree = parent.add("".join(label)) if parent else Tree("".join(label))
actions = self.actions.copy()
random.shuffle(actions)
await asyncio.gather(*(action.preview(parent=tree) for action in actions))
if not parent:
self.console.print(tree)
def __str__(self):
return (
f"ActionGroup(name={self.name!r}, actions={[a.name for a in self.actions]!r},"
f" inject_last_result={self.inject_last_result})"
)

156
falyx/action/base.py Normal file
View File

@ -0,0 +1,156 @@
# Falyx CLI Framework — (c) 2025 rtj.dev LLC — MIT Licensed
"""base.py
Core action system for Falyx.
This module defines the building blocks for executable actions and workflows,
providing a structured way to compose, execute, recover, and manage sequences of
operations.
All actions are callable and follow a unified signature:
result = action(*args, **kwargs)
Core guarantees:
- Full hook lifecycle support (before, on_success, on_error, after, on_teardown).
- Consistent timing and execution context tracking for each run.
- Unified, predictable result handling and error propagation.
- Optional last_result injection to enable flexible, data-driven workflows.
- Built-in support for retries, rollbacks, parallel groups, chaining, and fallback
recovery.
Key components:
- Action: wraps a function or coroutine into a standard executable unit.
- ChainedAction: runs actions sequentially, optionally injecting last results.
- ActionGroup: runs actions in parallel and gathers results.
- ProcessAction: executes CPU-bound functions in a separate process.
- LiteralInputAction: injects static values into workflows.
- FallbackAction: gracefully recovers from failures or missing data.
This design promotes clean, fault-tolerant, modular CLI and automation systems.
"""
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import Any, Callable
from rich.console import Console
from rich.tree import Tree
from falyx.context import SharedContext
from falyx.debug import register_debug_hooks
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
class BaseAction(ABC):
"""
Base class for actions. Actions can be simple functions or more
complex actions like `ChainedAction` or `ActionGroup`. They can also
be run independently or as part of Falyx.
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').
"""
def __init__(
self,
name: str,
*,
hooks: HookManager | None = None,
inject_last_result: bool = False,
inject_into: str = "last_result",
never_prompt: bool = False,
logging_hooks: bool = False,
) -> None:
self.name = name
self.hooks = hooks or HookManager()
self.is_retryable: bool = False
self.shared_context: SharedContext | None = None
self.inject_last_result: bool = inject_last_result
self.inject_into: str = inject_into
self._never_prompt: bool = never_prompt
self._skip_in_chain: bool = False
self.console = Console(color_system="auto")
self.options_manager: OptionsManager | None = None
if logging_hooks:
register_debug_hooks(self.hooks)
async def __call__(self, *args, **kwargs) -> Any:
return await self._run(*args, **kwargs)
@abstractmethod
async def _run(self, *args, **kwargs) -> Any:
raise NotImplementedError("_run must be implemented by subclasses")
@abstractmethod
async def preview(self, parent: Tree | None = None):
raise NotImplementedError("preview must be implemented by subclasses")
@abstractmethod
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.
"""
raise NotImplementedError("get_infer_target must be implemented by subclasses")
def set_options_manager(self, options_manager: OptionsManager) -> None:
self.options_manager = options_manager
def set_shared_context(self, shared_context: SharedContext) -> None:
self.shared_context = shared_context
def get_option(self, option_name: str, default: Any = None) -> Any:
"""
Resolve an option from the OptionsManager if present, otherwise use the fallback.
"""
if self.options_manager:
return self.options_manager.get(option_name, default)
return default
@property
def last_result(self) -> Any:
"""Return the last result from the shared context."""
if self.shared_context:
return self.shared_context.last_result()
return None
@property
def never_prompt(self) -> bool:
return self.get_option("never_prompt", self._never_prompt)
def prepare(
self, shared_context: SharedContext, options_manager: OptionsManager | None = None
) -> BaseAction:
"""
Prepare the action specifically for sequential (ChainedAction) execution.
Can be overridden for chain-specific logic.
"""
self.set_shared_context(shared_context)
if options_manager:
self.set_options_manager(options_manager)
return self
def _maybe_inject_last_result(self, kwargs: dict[str, Any]) -> dict[str, Any]:
if self.inject_last_result and self.shared_context:
key = self.inject_into
if key in kwargs:
logger.warning("[%s] Overriding '%s' with last_result", self.name, key)
kwargs = dict(kwargs)
kwargs[key] = self.shared_context.last_result()
return kwargs
def register_hooks_recursively(self, hook_type: HookType, hook: Hook):
"""Register a hook for all actions and sub-actions."""
self.hooks.register(hook_type, hook)
async def _write_stdout(self, data: str) -> None:
"""Override in subclasses that produce terminal output."""
def __repr__(self) -> str:
return str(self)

208
falyx/action/chained_action.py Normal file
View File

@ -0,0 +1,208 @@
from __future__ import annotations
from typing import Any, Callable
from rich.tree import Tree
from falyx.action.action import Action
from falyx.action.base import BaseAction
from falyx.action.fallback_action import FallbackAction
from falyx.action.literal_input_action import LiteralInputAction
from falyx.action.mixins import ActionListMixin
from falyx.context import ExecutionContext, SharedContext
from falyx.exceptions import EmptyChainError
from falyx.execution_registry import ExecutionRegistry as er
from falyx.hook_manager import Hook, HookManager, HookType
from falyx.logger import logger
from falyx.themes import OneColors
class ChainedAction(BaseAction, ActionListMixin):
"""
ChainedAction executes a sequence of actions one after another.
Features:
- Supports optional automatic last_result injection (auto_inject).
- Recovers from intermediate errors using FallbackAction if present.
- Rolls back all previously executed actions if a failure occurs.
- Handles literal values with LiteralInputAction.
Best used for defining robust, ordered workflows where each step can depend on
previous results.
Args:
name (str): Name of the chain.
actions (list): List of actions or literals to execute.
hooks (HookManager, optional): Hooks for lifecycle events.
inject_last_result (bool, optional): Whether to inject last results into kwargs
by default.
inject_into (str, optional): Key name for injection.
auto_inject (bool, optional): Auto-enable injection for subsequent actions.
return_list (bool, optional): Whether to return a list of all results. False
returns the last result.
"""
def __init__(
self,
name: str,
actions: list[BaseAction | Any] | None = None,
*,
hooks: HookManager | None = None,
inject_last_result: bool = False,
inject_into: str = "last_result",
auto_inject: bool = False,
return_list: bool = False,
) -> None:
super().__init__(
name,
hooks=hooks,
inject_last_result=inject_last_result,
inject_into=inject_into,
)
ActionListMixin.__init__(self)
self.auto_inject = auto_inject
self.return_list = return_list
if actions:
self.set_actions(actions)
def _wrap_if_needed(self, action: BaseAction | Any) -> BaseAction:
if isinstance(action, BaseAction):
return action
elif callable(action):
return Action(name=action.__name__, action=action)
else:
return LiteralInputAction(action)
def add_action(self, action: BaseAction | Any) -> None:
action = self._wrap_if_needed(action)
if self.actions and self.auto_inject and not action.inject_last_result:
action.inject_last_result = True
super().add_action(action)
if hasattr(action, "register_teardown") and callable(action.register_teardown):
action.register_teardown(self.hooks)
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, 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.")
shared_context = SharedContext(name=self.name, action=self)
if self.shared_context:
shared_context.add_result(self.shared_context.last_result())
updated_kwargs = self._maybe_inject_last_result(kwargs)
context = ExecutionContext(
name=self.name,
args=args,
kwargs=updated_kwargs,
action=self,
extra={"results": [], "rollback_stack": []},
shared_context=shared_context,
)
context.start_timer()
try:
await self.hooks.trigger(HookType.BEFORE, context)
for index, action in enumerate(self.actions):
if action._skip_in_chain:
logger.debug(
"[%s] Skipping consumed action '%s'", self.name, action.name
)
continue
shared_context.current_index = index
prepared = action.prepare(shared_context, self.options_manager)
try:
result = await prepared(*args, **updated_kwargs)
except Exception as error:
if index + 1 < len(self.actions) and isinstance(
self.actions[index + 1], FallbackAction
):
logger.warning(
"[%s] Fallback triggered: %s, recovering with fallback "
"'%s'.",
self.name,
error,
self.actions[index + 1].name,
)
shared_context.add_result(None)
context.extra["results"].append(None)
fallback = self.actions[index + 1].prepare(shared_context)
result = await fallback()
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)
all_results = context.extra["results"]
assert (
all_results
), f"[{self.name}] No results captured. Something seriously went wrong."
context.result = all_results if self.return_list else all_results[-1]
await self.hooks.trigger(HookType.ON_SUCCESS, context)
return context.result
except Exception as error:
context.exception = error
shared_context.add_error(shared_context.current_index, error)
await self._rollback(context.extra["rollback_stack"], *args, **kwargs)
await self.hooks.trigger(HookType.ON_ERROR, context)
raise
finally:
context.stop_timer()
await self.hooks.trigger(HookType.AFTER, context)
await self.hooks.trigger(HookType.ON_TEARDOWN, context)
er.record(context)
async def _rollback(self, rollback_stack, *args, **kwargs):
"""
Roll back all executed actions in reverse order.
Rollbacks run even if a fallback recovered from failure,
ensuring consistent undo of all side effects.
Actions without rollback handlers are skipped.
Args:
rollback_stack (list): Actions to roll back.
*args, **kwargs: Passed to rollback handlers.
"""
for action in reversed(rollback_stack):
rollback = getattr(action, "rollback", None)
if rollback:
try:
logger.warning("[%s] Rolling back...", action.name)
await action.rollback(*args, **kwargs)
except Exception as error:
logger.error("[%s] Rollback failed: %s", action.name, error)
def register_hooks_recursively(self, hook_type: HookType, hook: Hook):
"""Register a hook for all actions and sub-actions."""
self.hooks.register(hook_type, hook)
for action in self.actions:
action.register_hooks_recursively(hook_type, hook)
async def preview(self, parent: Tree | None = None):
label = [f"[{OneColors.CYAN_b}]⛓ ChainedAction[/] '{self.name}'"]
if self.inject_last_result:
label.append(f" [dim](injects '{self.inject_into}')[/dim]")
tree = parent.add("".join(label)) if parent else Tree("".join(label))
for action in self.actions:
await action.preview(parent=tree)
if not parent:
self.console.print(tree)
def __str__(self):
return (
f"ChainedAction(name={self.name!r}, "
f"actions={[a.name for a in self.actions]!r}, "
f"auto_inject={self.auto_inject}, return_list={self.return_list})"
)

49
falyx/action/fallback_action.py Normal file
View File

@ -0,0 +1,49 @@
from functools import cached_property
from typing import Any
from rich.tree import Tree
from falyx.action.action import Action
from falyx.themes import OneColors
class FallbackAction(Action):
"""
FallbackAction provides a default value if the previous action failed or
returned None.
It injects the last result and checks:
- If last_result is not None, it passes it through unchanged.
- If last_result is None (e.g., due to failure), it replaces it with a fallback value.
Used in ChainedAction pipelines to gracefully recover from errors or missing data.
When activated, it consumes the preceding error and allows the chain to continue
normally.
Args:
fallback (Any): The fallback value to use if last_result is None.
"""
def __init__(self, fallback: Any):
self._fallback = fallback
async def _fallback_logic(last_result):
return last_result if last_result is not None else fallback
super().__init__(name="Fallback", action=_fallback_logic, inject_last_result=True)
@cached_property
def fallback(self) -> Any:
"""Return the fallback value."""
return self._fallback
async def preview(self, parent: Tree | None = None):
label = [f"[{OneColors.LIGHT_RED}]🛟 Fallback[/] '{self.name}'"]
label.append(f" [dim](uses fallback = {repr(self.fallback)})[/dim]")
if parent:
parent.add("".join(label))
else:
self.console.print(Tree("".join(label)))
def __str__(self) -> str:
return f"FallbackAction(fallback={self.fallback!r})"

2
falyx/action/io_action.py
View File

@ -23,7 +23,7 @@ from typing import Any, Callable
from rich.tree import Tree
from falyx.action.action import BaseAction
from falyx.action.base import BaseAction
from falyx.context import ExecutionContext
from falyx.exceptions import FalyxError
from falyx.execution_registry import ExecutionRegistry as er

47
falyx/action/literal_input_action.py Normal file
View File

@ -0,0 +1,47 @@
from __future__ import annotations
from functools import cached_property
from typing import Any
from rich.tree import Tree
from falyx.action.action import Action
from falyx.themes import OneColors
class LiteralInputAction(Action):
"""
LiteralInputAction injects a static value into a ChainedAction.
This allows embedding hardcoded values mid-pipeline, useful when:
- Providing default or fallback inputs.
- Starting a pipeline with a fixed input.
- Supplying missing context manually.
Args:
value (Any): The static value to inject.
"""
def __init__(self, value: Any):
self._value = value
async def literal(*_, **__):
return value
super().__init__("Input", literal)
@cached_property
def value(self) -> Any:
"""Return the literal value."""
return self._value
async def preview(self, parent: Tree | None = None):
label = [f"[{OneColors.LIGHT_YELLOW}]📥 LiteralInput[/] '{self.name}'"]
label.append(f" [dim](value = {repr(self.value)})[/dim]")
if parent:
parent.add("".join(label))
else:
self.console.print(Tree("".join(label)))
def __str__(self) -> str:
return f"LiteralInputAction(value={self.value!r})"

2
falyx/action/menu_action.py
View File

@ -7,7 +7,7 @@ from rich.console import Console
from rich.table import Table
from rich.tree import Tree
from falyx.action.action import BaseAction
from falyx.action.base import BaseAction
from falyx.context import ExecutionContext
from falyx.execution_registry import ExecutionRegistry as er
from falyx.hook_manager import HookType

33
falyx/action/mixins.py Normal file
View File

@ -0,0 +1,33 @@
from falyx.action.base import BaseAction
class ActionListMixin:
"""Mixin for managing a list of actions."""
def __init__(self) -> None:
self.actions: list[BaseAction] = []
def set_actions(self, actions: list[BaseAction]) -> None:
"""Replaces the current action list with a new one."""
self.actions.clear()
for action in actions:
self.add_action(action)
def add_action(self, action: BaseAction) -> None:
"""Adds an action to the list."""
self.actions.append(action)
def remove_action(self, name: str) -> None:
"""Removes an action by name."""
self.actions = [action for action in self.actions if action.name != name]
def has_action(self, name: str) -> bool:
"""Checks if an action with the given name exists."""
return any(action.name == name for action in self.actions)
def get_action(self, name: str) -> BaseAction | None:
"""Retrieves an action by name."""
for action in self.actions:
if action.name == name:
return action
return None

128
falyx/action/process_action.py Normal file
View File

@ -0,0 +1,128 @@
from __future__ import annotations
import asyncio
from concurrent.futures import ProcessPoolExecutor
from functools import partial
from typing import Any, Callable
from rich.tree import Tree
from falyx.action.base import BaseAction
from falyx.context import ExecutionContext
from falyx.execution_registry import ExecutionRegistry as er
from falyx.hook_manager import HookManager, HookType
from falyx.themes import OneColors
class ProcessAction(BaseAction):
"""
ProcessAction runs a function in a separate process using ProcessPoolExecutor.
Features:
- Executes CPU-bound or blocking tasks without blocking the main event loop.
- Supports last_result injection into the subprocess.
- Validates that last_result is pickleable when injection is enabled.
Args:
name (str): Name of the action.
func (Callable): Function to execute in a new process.
args (tuple, optional): Positional arguments.
kwargs (dict, optional): Keyword arguments.
hooks (HookManager, optional): Hook manager for lifecycle events.
executor (ProcessPoolExecutor, optional): Custom executor if desired.
inject_last_result (bool, optional): Inject last result into the function.
inject_into (str, optional): Name of the injected key.
"""
def __init__(
self,
name: str,
action: Callable[..., Any],
*,
args: tuple = (),
kwargs: dict[str, Any] | None = None,
hooks: HookManager | None = None,
executor: ProcessPoolExecutor | None = None,
inject_last_result: bool = False,
inject_into: str = "last_result",
):
super().__init__(
name,
hooks=hooks,
inject_last_result=inject_last_result,
inject_into=inject_into,
)
self.action = action
self.args = args
self.kwargs = kwargs or {}
self.executor = executor or ProcessPoolExecutor()
self.is_retryable = True
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:
last_result = self.shared_context.last_result()
if not self._validate_pickleable(last_result):
raise ValueError(
f"Cannot inject last result into {self.name}: "
f"last result is not pickleable."
)
combined_args = args + self.args
combined_kwargs = self._maybe_inject_last_result({**self.kwargs, **kwargs})
context = ExecutionContext(
name=self.name,
args=combined_args,
kwargs=combined_kwargs,
action=self,
)
loop = asyncio.get_running_loop()
context.start_timer()
try:
await self.hooks.trigger(HookType.BEFORE, context)
result = await loop.run_in_executor(
self.executor, partial(self.action, *combined_args, **combined_kwargs)
)
context.result = result
await self.hooks.trigger(HookType.ON_SUCCESS, context)
return result
except Exception as error:
context.exception = error
await self.hooks.trigger(HookType.ON_ERROR, context)
if context.result is not None:
return context.result
raise
finally:
context.stop_timer()
await self.hooks.trigger(HookType.AFTER, context)
await self.hooks.trigger(HookType.ON_TEARDOWN, context)
er.record(context)
def _validate_pickleable(self, obj: Any) -> bool:
try:
import pickle
pickle.dumps(obj)
return True
except (pickle.PicklingError, TypeError):
return False
async def preview(self, parent: Tree | None = None):
label = [
f"[{OneColors.DARK_YELLOW_b}]🧠 ProcessAction (new process)[/] '{self.name}'"
]
if self.inject_last_result:
label.append(f" [dim](injects '{self.inject_into}')[/dim]")
if parent:
parent.add("".join(label))
else:
self.console.print(Tree("".join(label)))
def __str__(self) -> str:
return (
f"ProcessAction(name={self.name!r}, "
f"action={getattr(self.action, '__name__', repr(self.action))}, "
f"args={self.args!r}, kwargs={self.kwargs!r})"
)

2
falyx/action/prompt_menu_action.py
View File

@ -7,7 +7,7 @@ from prompt_toolkit.formatted_text import FormattedText, merge_formatted_text
from rich.console import Console
from rich.tree import Tree
from falyx.action.action import BaseAction
from falyx.action.base import BaseAction
from falyx.context import ExecutionContext
from falyx.execution_registry import ExecutionRegistry as er
from falyx.hook_manager import HookType

2
falyx/action/select_file_action.py
View File

@ -14,7 +14,7 @@ from prompt_toolkit import PromptSession
from rich.console import Console
from rich.tree import Tree
from falyx.action.action import BaseAction
from falyx.action.base import BaseAction
from falyx.action.types import FileReturnType
from falyx.context import ExecutionContext
from falyx.execution_registry import ExecutionRegistry as er

3
falyx/action/selection_action.py
View File

@ -1,13 +1,12 @@
# Falyx CLI Framework — (c) 2025 rtj.dev LLC — MIT Licensed
"""selection_action.py"""
from copy import copy
from typing import Any
from prompt_toolkit import PromptSession
from rich.console import Console
from rich.tree import Tree
from falyx.action.action import BaseAction
from falyx.action.base import BaseAction
from falyx.action.types import SelectionReturnType
from falyx.context import ExecutionContext
from falyx.execution_registry import ExecutionRegistry as er

2
falyx/action/user_input_action.py
View File

@ -3,7 +3,7 @@ from prompt_toolkit.validation import Validator
from rich.console import Console
from rich.tree import Tree
from falyx.action import BaseAction
from falyx.action.base import BaseAction
from falyx.context import ExecutionContext
from falyx.execution_registry import ExecutionRegistry as er
from falyx.hook_manager import HookType

3
falyx/command.py
View File

@ -26,7 +26,8 @@ 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, BaseAction
from falyx.action.action import Action
from falyx.action.base import BaseAction
from falyx.context import ExecutionContext
from falyx.debug import register_debug_hooks
from falyx.execution_registry import ExecutionRegistry as er

3
falyx/config.py
View File

@ -13,7 +13,8 @@ import yaml
from pydantic import BaseModel, Field, field_validator, model_validator
from rich.console import Console
from falyx.action.action import Action, BaseAction
from falyx.action.action import Action
from falyx.action.base import BaseAction
from falyx.command import Command
from falyx.falyx import Falyx
from falyx.logger import logger

28
falyx/falyx.py
View File

@ -42,7 +42,8 @@ from rich.console import Console
from rich.markdown import Markdown
from rich.table import Table
from falyx.action.action import Action, BaseAction
from falyx.action.action import Action
from falyx.action.base import BaseAction
from falyx.bottom_bar import BottomBar
from falyx.command import Command
from falyx.context import ExecutionContext
@ -82,7 +83,7 @@ class CommandValidator(Validator):
self.falyx = falyx
self.error_message = error_message
def validate(self, document) -> None:
def validate(self, _) -> None:
pass
async def validate_async(self, document) -> None:
@ -449,7 +450,7 @@ class Falyx:
validator=CommandValidator(self, self._get_validator_error_message()),
bottom_toolbar=self._get_bottom_bar_render(),
key_bindings=self.key_bindings,
validate_while_typing=False,
validate_while_typing=True,
)
return self._prompt_session
@ -761,7 +762,7 @@ class Falyx:
is_preview = False
choice = "?"
elif is_preview and not choice:
# No help command enabled
# No help (list) command enabled
if not from_validate:
self.console.print(
f"[{OneColors.DARK_RED}]❌ You must enter a command for preview mode."
@ -781,12 +782,9 @@ class Falyx:
)
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()
self.console.print(f"[{OneColors.DARK_RED}]❌ [{choice}]: {error}")
else:
raise ValidationError(
message=str(error), cursor_position=len(raw_choices)
)
@ -809,11 +807,21 @@ class Falyx:
for match in fuzzy_matches:
cmd = name_map[match]
self.console.print(f" • [bold]{match}[/] → {cmd.description}")
else:
raise ValidationError(
message=f"Unknown command '{choice}'. Did you mean: "
f"{', '.join(fuzzy_matches)}?",
cursor_position=len(raw_choices),
)
else:
if not from_validate:
self.console.print(
f"[{OneColors.LIGHT_YELLOW}]⚠️ Unknown command '{choice}'[/]"
)
raise ValidationError(
message=f"Unknown command '{choice}'.",
cursor_position=len(raw_choices),
)
return is_preview, None, args, kwargs
def _create_context(self, selected_command: Command) -> ExecutionContext:
@ -974,7 +982,7 @@ class Falyx:
async def menu(self) -> None:
"""Runs the menu and handles user input."""
logger.info("Running menu: %s", self.get_title())
logger.info("Starting menu: %s", self.get_title())
self.debug_hooks()
if self.welcome_message:
self.print_message(self.welcome_message)

2
falyx/menu.py
View File

@ -4,7 +4,7 @@ from dataclasses import dataclass
from prompt_toolkit.formatted_text import FormattedText
from falyx.action import BaseAction
from falyx.action.base import BaseAction
from falyx.signals import BackSignal, QuitSignal
from falyx.themes import OneColors
from falyx.utils import CaseInsensitiveDict

173
falyx/parsers/argparse.py
View File

@ -10,6 +10,7 @@ from rich.console import Console
from rich.markup import escape
from rich.text import Text
from falyx.action.base import BaseAction
from falyx.exceptions import CommandArgumentError
from falyx.signals import HelpSignal
@ -17,6 +18,7 @@ from falyx.signals import HelpSignal
class ArgumentAction(Enum):
"""Defines the action to be taken when the argument is encountered."""
ACTION = "action"
STORE = "store"
STORE_TRUE = "store_true"
STORE_FALSE = "store_false"
@ -51,6 +53,7 @@ class Argument:
help: str = "" # Help text for the argument
nargs: int | str | None = None # int, '?', '*', '+', None
positional: bool = False # True if no leading - or -- in flags
resolver: BaseAction | None = None # Action object for the argument
def get_positional_text(self) -> str:
"""Get the positional text for the argument."""
@ -104,6 +107,8 @@ class Argument:
and self.required == other.required
and self.nargs == other.nargs
and self.positional == other.positional
and self.default == other.default
and self.help == other.help
)
def __hash__(self) -> int:
@ -117,6 +122,8 @@ class Argument:
self.required,
self.nargs,
self.positional,
self.default,
self.help,
)
)
@ -220,6 +227,12 @@ class CommandArgumentParser:
if required:
return True
if positional:
assert (
nargs is None
or isinstance(nargs, int)
or isinstance(nargs, str)
and nargs in ("+", "*", "?")
), f"Invalid nargs value: {nargs}"
if isinstance(nargs, int):
return nargs > 0
elif isinstance(nargs, str):
@ -228,7 +241,7 @@ class CommandArgumentParser:
elif nargs in ("*", "?"):
return False
else:
raise CommandArgumentError(f"Invalid nargs value: {nargs}")
return True
return required
@ -247,7 +260,7 @@ class CommandArgumentParser:
)
return None
if nargs is None:
nargs = 1
return None
allowed_nargs = ("?", "*", "+")
if isinstance(nargs, int):
if nargs <= 0:
@ -308,6 +321,23 @@ class CommandArgumentParser:
f"Default list value {default!r} for '{dest}' cannot be coerced to {expected_type.__name__}"
)
def _validate_resolver(
self, action: ArgumentAction, resolver: BaseAction | None
) -> BaseAction | None:
"""Validate the action object."""
if action != ArgumentAction.ACTION and resolver is None:
return None
elif action == ArgumentAction.ACTION and resolver is None:
raise CommandArgumentError("resolver must be provided for ACTION action")
elif action != ArgumentAction.ACTION and resolver is not None:
raise CommandArgumentError(
f"resolver should not be provided for action {action}"
)
if not isinstance(resolver, BaseAction):
raise CommandArgumentError("resolver must be an instance of BaseAction")
return resolver
def _validate_action(
self, action: ArgumentAction | str, positional: bool
) -> ArgumentAction:
@ -347,6 +377,8 @@ class CommandArgumentParser:
return 0
elif action in (ArgumentAction.APPEND, ArgumentAction.EXTEND):
return []
elif isinstance(nargs, int):
return []
elif nargs in ("+", "*"):
return []
else:
@ -380,8 +412,15 @@ class CommandArgumentParser:
required: bool = False,
help: str = "",
dest: str | None = None,
resolver: BaseAction | None = None,
) -> None:
"""Add an argument to the parser.
For `ArgumentAction.ACTION`, `nargs` and `type` determine how many and what kind
of inputs are passed to the `resolver`.
The return value of the `resolver` is used directly (no type coercion is applied).
Validation, structure, and post-processing should be handled within the `resolver`.
Args:
name or flags: Either a name or prefixed flags (e.g. 'faylx', '-f', '--falyx').
action: The action to be taken when the argument is encountered.
@ -392,6 +431,7 @@ class CommandArgumentParser:
required: Whether or not the argument is required.
help: A brief description of the argument.
dest: The name of the attribute to be added to the object returned by parse_args().
resolver: A BaseAction called with optional nargs specified parsed arguments.
"""
expected_type = type
self._validate_flags(flags)
@ -403,8 +443,8 @@ class CommandArgumentParser:
"Merging multiple arguments into the same dest (e.g. positional + flagged) "
"is not supported. Define a unique 'dest' for each argument."
)
self._dest_set.add(dest)
action = self._validate_action(action, positional)
resolver = self._validate_resolver(action, resolver)
nargs = self._validate_nargs(nargs, action)
default = self._resolve_default(default, action, nargs)
if (
@ -432,6 +472,7 @@ class CommandArgumentParser:
help=help,
nargs=nargs,
positional=positional,
resolver=resolver,
)
for flag in flags:
if flag in self._flag_map:
@ -439,7 +480,9 @@ class CommandArgumentParser:
raise CommandArgumentError(
f"Flag '{flag}' is already used by argument '{existing.dest}'"
)
for flag in flags:
self._flag_map[flag] = argument
self._dest_set.add(dest)
self._arguments.append(argument)
if positional:
self._positional.append(argument)
@ -462,6 +505,8 @@ class CommandArgumentParser:
"required": arg.required,
"nargs": arg.nargs,
"positional": arg.positional,
"default": arg.default,
"help": arg.help,
}
)
return defs
@ -469,14 +514,17 @@ class CommandArgumentParser:
def _consume_nargs(
self, args: list[str], start: int, spec: Argument
) -> tuple[list[str], int]:
assert (
spec.nargs is None
or isinstance(spec.nargs, int)
or isinstance(spec.nargs, str)
and spec.nargs in ("+", "*", "?")
), f"Invalid nargs value: {spec.nargs}"
values = []
i = start
if isinstance(spec.nargs, int):
values = args[i : i + spec.nargs]
return values, i + spec.nargs
elif spec.nargs is None:
values = [args[i]]
return values, i + 1
elif spec.nargs == "+":
if i >= len(args):
raise CommandArgumentError(
@ -496,10 +544,13 @@ class CommandArgumentParser:
if i < len(args) and not args[i].startswith("-"):
return [args[i]], i + 1
return [], i
else:
elif spec.nargs is None:
if i < len(args) and not args[i].startswith("-"):
return [args[i]], i + 1
return [], i
assert False, "Invalid nargs value: shouldn't happen"
def _consume_all_positional_args(
async def _consume_all_positional_args(
self,
args: list[str],
result: dict[str, Any],
@ -519,18 +570,22 @@ class CommandArgumentParser:
remaining = len(args) - i
min_required = 0
for next_spec in positional_args[j + 1 :]:
if isinstance(next_spec.nargs, int):
min_required += next_spec.nargs
elif next_spec.nargs is None:
assert (
next_spec.nargs is None
or isinstance(next_spec.nargs, int)
or isinstance(next_spec.nargs, str)
and next_spec.nargs in ("+", "*", "?")
), f"Invalid nargs value: {spec.nargs}"
if next_spec.nargs is None:
min_required += 1
elif isinstance(next_spec.nargs, int):
min_required += next_spec.nargs
elif next_spec.nargs == "+":
min_required += 1
elif next_spec.nargs == "?":
min_required += 0
elif next_spec.nargs == "*":
min_required += 0
else:
assert False, "Invalid nargs value: shouldn't happen"
slice_args = args[i:] if is_last else args[i : i + (remaining - min_required)]
values, new_i = self._consume_nargs(slice_args, 0, spec)
@ -542,10 +597,19 @@ class CommandArgumentParser:
raise CommandArgumentError(
f"Invalid value for '{spec.dest}': expected {spec.type.__name__}"
)
if spec.action == ArgumentAction.APPEND:
if spec.action == ArgumentAction.ACTION:
assert isinstance(
spec.resolver, BaseAction
), "resolver should be an instance of BaseAction"
try:
result[spec.dest] = await spec.resolver(*typed)
except Exception as error:
raise CommandArgumentError(
f"[{spec.dest}] Action failed: {error}"
) from error
elif spec.action == ArgumentAction.APPEND:
assert result.get(spec.dest) is not None, "dest should not be None"
if spec.nargs in (None, 1):
if spec.nargs is None:
result[spec.dest].append(typed[0])
else:
result[spec.dest].append(typed)
@ -565,6 +629,23 @@ class CommandArgumentParser:
return i
def _expand_posix_bundling(self, args: list[str]) -> list[str]:
"""Expand POSIX-style bundled arguments into separate arguments."""
expanded = []
for token in args:
if token.startswith("-") and not token.startswith("--") and len(token) > 2:
# POSIX bundle
# e.g. -abc -> -a -b -c
for char in token[1:]:
flag = f"-{char}"
arg = self._flag_map.get(flag)
if not arg:
raise CommandArgumentError(f"Unrecognized option: {flag}")
expanded.append(flag)
else:
expanded.append(token)
return expanded
async def parse_args(
self, args: list[str] | None = None, from_validate: bool = False
) -> dict[str, Any]:
@ -572,11 +653,13 @@ class CommandArgumentParser:
if args is None:
args = []
args = self._expand_posix_bundling(args)
result = {arg.dest: deepcopy(arg.default) for arg in self._arguments}
positional_args = [arg for arg in self._arguments if arg.positional]
consumed_positional_indices: set[int] = set()
consumed_indices: set[int] = set()
i = 0
while i < len(args):
token = args[i]
@ -588,6 +671,25 @@ class CommandArgumentParser:
if not from_validate:
self.render_help()
raise HelpSignal()
elif action == ArgumentAction.ACTION:
assert isinstance(
spec.resolver, BaseAction
), "resolver should be an instance of BaseAction"
values, new_i = self._consume_nargs(args, i + 1, spec)
try:
typed_values = [spec.type(value) for value in values]
except ValueError:
raise CommandArgumentError(
f"Invalid value for '{spec.dest}': expected {spec.type.__name__}"
)
try:
result[spec.dest] = await spec.resolver(*typed_values)
except Exception as error:
raise CommandArgumentError(
f"[{spec.dest}] Action failed: {error}"
) from error
consumed_indices.update(range(i, new_i))
i = new_i
elif action == ArgumentAction.STORE_TRUE:
result[spec.dest] = True
consumed_indices.add(i)
@ -609,13 +711,8 @@ class CommandArgumentParser:
raise CommandArgumentError(
f"Invalid value for '{spec.dest}': expected {spec.type.__name__}"
)
if spec.nargs in (None, 1):
try:
if spec.nargs is None:
result[spec.dest].append(spec.type(values[0]))
except ValueError:
raise CommandArgumentError(
f"Invalid value for '{spec.dest}': expected {spec.type.__name__}"
)
else:
result[spec.dest].append(typed_values)
consumed_indices.update(range(i, new_i))
@ -640,6 +737,10 @@ class CommandArgumentParser:
raise CommandArgumentError(
f"Invalid value for '{spec.dest}': expected {spec.type.__name__}"
)
if not typed_values and spec.nargs not in ("*", "?"):
raise CommandArgumentError(
f"Expected at least one value for '{spec.dest}'"
)
if (
spec.nargs in (None, 1, "?")
and spec.action != ArgumentAction.APPEND
@ -651,6 +752,9 @@ class CommandArgumentParser:
result[spec.dest] = typed_values
consumed_indices.update(range(i, new_i))
i = new_i
elif token.startswith("-"):
# Handle unrecognized option
raise CommandArgumentError(f"Unrecognized flag: {token}")
else:
# Get the next flagged argument index if it exists
next_flagged_index = -1
@ -660,8 +764,7 @@ class CommandArgumentParser:
break
if next_flagged_index == -1:
next_flagged_index = len(args)
args_consumed = self._consume_all_positional_args(
args_consumed = await self._consume_all_positional_args(
args[i:next_flagged_index],
result,
positional_args,
@ -681,26 +784,22 @@ class CommandArgumentParser:
f"Invalid value for {spec.dest}: must be one of {spec.choices}"
)
if spec.action == ArgumentAction.ACTION:
continue
if isinstance(spec.nargs, int) and spec.nargs > 1:
if not isinstance(result.get(spec.dest), list):
raise CommandArgumentError(
f"Invalid value for {spec.dest}: expected a list"
)
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
if spec.action == ArgumentAction.APPEND:
if not isinstance(result[spec.dest], list):
raise CommandArgumentError(
f"Invalid value for {spec.dest}: expected a list"
)
for group in result[spec.dest]:
if len(group) % spec.nargs != 0:
raise CommandArgumentError(
f"Invalid number of values for {spec.dest}: expected a multiple of {spec.nargs}"
)
elif spec.action == ArgumentAction.EXTEND:
if not isinstance(result[spec.dest], list):
raise CommandArgumentError(
f"Invalid value for {spec.dest}: expected a list"
)
if len(result[spec.dest]) % spec.nargs != 0:
raise CommandArgumentError(
f"Invalid number of values for {spec.dest}: expected a multiple of {spec.nargs}"

2
falyx/parsers/utils.py
View File

@ -1,5 +1,6 @@
from typing import Any
from falyx.action.base import BaseAction
from falyx.logger import logger
from falyx.parsers.signature import infer_args_from_func
@ -8,7 +9,6 @@ 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:

2
falyx/protocols.py
View File

@ -4,7 +4,7 @@ from __future__ import annotations
from typing import Any, Awaitable, Protocol, runtime_checkable
from falyx.action.action import BaseAction
from falyx.action.base import BaseAction
@runtime_checkable

3
falyx/retry_utils.py
View File

@ -1,6 +1,7 @@
# Falyx CLI Framework — (c) 2025 rtj.dev LLC — MIT Licensed
"""retry_utils.py"""
from falyx.action.action import Action, BaseAction
from falyx.action.action import Action
from falyx.action.base import BaseAction
from falyx.hook_manager import HookType
from falyx.retry import RetryHandler, RetryPolicy

2
falyx/utils.py
View File

@ -184,7 +184,7 @@ def setup_logging(
console_handler.setLevel(console_log_level)
root.addHandler(console_handler)
file_handler = logging.FileHandler(log_filename)
file_handler = logging.FileHandler(log_filename, "a", "UTF-8")
file_handler.setLevel(file_log_level)
if json_log_to_file:
file_handler.setFormatter(

2
falyx/version.py
View File

@ -1 +1 @@
__version__ = "0.1.37"
__version__ = "0.1.38"

2
pyproject.toml
View File

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

124
tests/test_command_argument_parser.py
View File

@ -345,22 +345,24 @@ def test_add_argument_choices_invalid():
def test_add_argument_bad_nargs():
parser = CommandArgumentParser()
# ❌ Invalid nargs value
with pytest.raises(CommandArgumentError):
parser.add_argument("--falyx", nargs="invalid")
# ❌ Invalid nargs type
with pytest.raises(CommandArgumentError):
parser.add_argument("--falyx", nargs=123)
parser.add_argument("--foo", nargs="123")
# ❌ Invalid nargs type
with pytest.raises(CommandArgumentError):
parser.add_argument("--falyx", nargs=None)
parser.add_argument("--foo", nargs=[1, 2])
with pytest.raises(CommandArgumentError):
parser.add_argument("--too", action="count", nargs=5)
with pytest.raises(CommandArgumentError):
parser.add_argument("falyx", action="store_true", nargs=5)
def test_add_argument_nargs():
parser = CommandArgumentParser()
# ✅ Valid nargs value
parser.add_argument("--falyx", nargs=2)
arg = parser._arguments[-1]
assert arg.dest == "falyx"
@ -398,8 +400,10 @@ async def test_parse_args_nargs():
parser = CommandArgumentParser()
parser.add_argument("files", nargs="+", type=str)
parser.add_argument("mode", nargs=1)
parser.add_argument("--action", action="store_true")
args = await parser.parse_args(["a", "b", "c"])
args = await parser.parse_args(["a", "b", "c", "--action"])
args = await parser.parse_args(["--action", "a", "b", "c"])
assert args["files"] == ["a", "b"]
assert args["mode"] == "c"
@ -517,6 +521,15 @@ async def test_parse_args_nargs_multiple_positional():
await parser.parse_args([])
@pytest.mark.asyncio
async def test_parse_args_nargs_none():
parser = CommandArgumentParser()
parser.add_argument("numbers", type=int)
parser.add_argument("mode")
await parser.parse_args(["1", "2"])
@pytest.mark.asyncio
async def test_parse_args_nargs_invalid_positional_arguments():
parser = CommandArgumentParser()
@ -542,20 +555,78 @@ async def test_parse_args_append():
assert args["numbers"] == []
@pytest.mark.asyncio
async def test_parse_args_nargs_int_append():
parser = CommandArgumentParser()
parser.add_argument("--numbers", action=ArgumentAction.APPEND, type=int, nargs=1)
args = await parser.parse_args(["--numbers", "1", "--numbers", "2", "--numbers", "3"])
assert args["numbers"] == [[1], [2], [3]]
args = await parser.parse_args(["--numbers", "1"])
assert args["numbers"] == [[1]]
args = await parser.parse_args([])
assert args["numbers"] == []
@pytest.mark.asyncio
async def test_parse_args_nargs_append():
parser = CommandArgumentParser()
parser.add_argument("numbers", action=ArgumentAction.APPEND, type=int, nargs="*")
parser.add_argument("--mode")
args = await parser.parse_args(["1"])
assert args["numbers"] == [[1]]
args = await parser.parse_args(["1", "2", "3", "--mode", "numbers", "4", "5"])
assert args["numbers"] == [[1, 2, 3], [4, 5]]
assert args["mode"] == "numbers"
args = await parser.parse_args(["1", "2", "3"])
assert args["numbers"] == [[1, 2, 3]]
args = await parser.parse_args([])
assert args["numbers"] == []
@pytest.mark.asyncio
async def test_parse_args_int_optional_append():
parser = CommandArgumentParser()
parser.add_argument("numbers", action=ArgumentAction.APPEND, type=int)
args = await parser.parse_args(["1"])
assert args["numbers"] == [1]
@pytest.mark.asyncio
async def test_parse_args_int_optional_append_multiple_values():
parser = CommandArgumentParser()
parser.add_argument("numbers", action=ArgumentAction.APPEND, type=int)
with pytest.raises(CommandArgumentError):
await parser.parse_args(["1", "2"])
@pytest.mark.asyncio
async def test_parse_args_nargs_int_positional_append():
parser = CommandArgumentParser()
parser.add_argument("numbers", action=ArgumentAction.APPEND, type=int, nargs=1)
args = await parser.parse_args(["1"])
assert args["numbers"] == [[1]]
args = await parser.parse_args([])
assert args["numbers"] == []
with pytest.raises(CommandArgumentError):
await parser.parse_args(["1", "2", "3"])
parser2 = CommandArgumentParser()
parser2.add_argument("numbers", action=ArgumentAction.APPEND, type=int, nargs=2)
args = await parser2.parse_args(["1", "2"])
assert args["numbers"] == [[1, 2]]
with pytest.raises(CommandArgumentError):
await parser2.parse_args(["1", "2", "3"])
@pytest.mark.asyncio
@ -575,6 +646,9 @@ async def test_append_groups_nargs():
parsed = await cap.parse_args(["--item", "a", "b", "--item", "c", "d"])
assert parsed["item"] == [["a", "b"], ["c", "d"]]
with pytest.raises(CommandArgumentError):
await cap.parse_args(["--item", "a", "b", "--item", "c"])
@pytest.mark.asyncio
async def test_extend_flattened():
@ -720,3 +794,35 @@ async def test_extend_positional_nargs():
with pytest.raises(CommandArgumentError):
await parser.parse_args([])
def test_command_argument_parser_equality():
parser1 = CommandArgumentParser()
parser2 = CommandArgumentParser()
parser1.add_argument("--foo", type=str)
parser2.add_argument("--foo", type=str)
assert parser1 == parser2
parser1.add_argument("--bar", type=int)
assert parser1 != parser2
parser2.add_argument("--bar", type=int)
assert parser1 == parser2
assert parser1 != "not a parser"
assert parser1 is not None
assert parser1 != object()
assert parser1.to_definition_list() == parser2.to_definition_list()
assert hash(parser1) == hash(parser2)
@pytest.mark.asyncio
async def test_render_help():
parser = CommandArgumentParser()
parser.add_argument("--foo", type=str, help="Foo help")
parser.add_argument("--bar", action=ArgumentAction.APPEND, type=str, help="Bar help")
assert parser.render_help() is None

227
tests/test_parsers/test_action.py Normal file
View File

@ -0,0 +1,227 @@
import pytest
from falyx.action import Action, SelectionAction
from falyx.exceptions import CommandArgumentError
from falyx.parsers import ArgumentAction, CommandArgumentParser
def test_add_argument():
"""Test the add_argument method."""
parser = CommandArgumentParser()
action = Action("test_action", lambda: "value")
parser.add_argument(
"test", action=ArgumentAction.ACTION, help="Test argument", resolver=action
)
with pytest.raises(CommandArgumentError):
parser.add_argument("test1", action=ArgumentAction.ACTION, help="Test argument")
with pytest.raises(CommandArgumentError):
parser.add_argument(
"test2",
action=ArgumentAction.ACTION,
help="Test argument",
resolver="Not an action",
)
@pytest.mark.asyncio
async def test_falyx_actions():
"""Test the Falyx actions."""
parser = CommandArgumentParser()
action = Action("test_action", lambda: "value")
parser.add_argument(
"-a",
"--alpha",
action=ArgumentAction.ACTION,
resolver=action,
help="Alpha option",
)
# Test valid cases
args = await parser.parse_args(["-a"])
assert args["alpha"] == "value"
@pytest.mark.asyncio
async def test_action_basic():
parser = CommandArgumentParser()
action = Action("hello", lambda: "hi")
parser.add_argument("--greet", action=ArgumentAction.ACTION, resolver=action)
args = await parser.parse_args(["--greet"])
assert args["greet"] == "hi"
@pytest.mark.asyncio
async def test_action_with_nargs():
parser = CommandArgumentParser()
def multiply(a, b):
return int(a) * int(b)
action = Action("multiply", multiply)
parser.add_argument("--mul", action=ArgumentAction.ACTION, resolver=action, nargs=2)
args = await parser.parse_args(["--mul", "3", "4"])
assert args["mul"] == 12
@pytest.mark.asyncio
async def test_action_with_nargs_positional():
parser = CommandArgumentParser()
def multiply(a, b):
return int(a) * int(b)
action = Action("multiply", multiply)
parser.add_argument("mul", action=ArgumentAction.ACTION, resolver=action, nargs=2)
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"])
@pytest.mark.asyncio
async def test_action_with_nargs_positional_int():
parser = CommandArgumentParser()
def multiply(a, b):
return a * b
action = Action("multiply", multiply)
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(["abc", "3"])
@pytest.mark.asyncio
async def test_action_with_nargs_type():
parser = CommandArgumentParser()
def multiply(a, b):
return a * b
action = Action("multiply", multiply)
parser.add_argument(
"--mul", action=ArgumentAction.ACTION, resolver=action, nargs=2, type=int
)
args = await parser.parse_args(["--mul", "3", "4"])
assert args["mul"] == 12
with pytest.raises(CommandArgumentError):
await parser.parse_args(["--mul", "abc", "3"])
@pytest.mark.asyncio
async def test_action_with_custom_type():
parser = CommandArgumentParser()
def upcase(s):
return s.upper()
action = Action("upcase", upcase)
parser.add_argument("--word", action=ArgumentAction.ACTION, resolver=action, type=str)
args = await parser.parse_args(["--word", "hello"])
assert args["word"] == "HELLO"
@pytest.mark.asyncio
async def test_action_with_nargs_star():
parser = CommandArgumentParser()
def joiner(*args):
return "-".join(args)
action = Action("join", joiner)
parser.add_argument(
"--tags", action=ArgumentAction.ACTION, resolver=action, nargs="*"
)
args = await parser.parse_args(["--tags", "a", "b", "c"])
assert args["tags"] == "a-b-c"
@pytest.mark.asyncio
async def test_action_nargs_plus_missing():
parser = CommandArgumentParser()
action = Action("noop", lambda *args: args)
parser.add_argument("--x", action=ArgumentAction.ACTION, resolver=action, nargs="+")
with pytest.raises(CommandArgumentError):
await parser.parse_args(["--x"])
@pytest.mark.asyncio
async def test_action_with_default():
parser = CommandArgumentParser()
action = Action("default", lambda value: value)
parser.add_argument(
"--default",
action=ArgumentAction.ACTION,
resolver=action,
default="default_value",
)
args = await parser.parse_args([])
assert args["default"] == "default_value"
@pytest.mark.asyncio
async def test_action_with_default_and_value():
parser = CommandArgumentParser()
action = Action("default", lambda value: value)
parser.add_argument(
"--default",
action=ArgumentAction.ACTION,
resolver=action,
default="default_value",
)
args = await parser.parse_args(["--default", "new_value"])
assert args["default"] == "new_value"
@pytest.mark.asyncio
async def test_action_with_default_and_value_not():
parser = CommandArgumentParser()
action = Action("default", lambda: "default_value")
parser.add_argument(
"--default",
action=ArgumentAction.ACTION,
resolver=action,
default="default_value",
)
with pytest.raises(CommandArgumentError):
await parser.parse_args(["--default", "new_value"])
@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)
with pytest.raises(CommandArgumentError):
await parser.parse_args([])
with pytest.raises(CommandArgumentError):
await parser.parse_args(["be"])
# @pytest.mark.asyncio
# async def test_selection_action():
# parser = CommandArgumentParser()
# action = SelectionAction("select", selections=["a", "b", "c"])
# parser.add_argument("--select", action=ArgumentAction.ACTION, resolver=action)
# args = await parser.parse_args(["--select"])

90
tests/test_parsers/test_argument.py Normal file
View File

@ -0,0 +1,90 @@
import pytest
from falyx.parsers import Argument, ArgumentAction
def test_positional_text_with_choices():
arg = Argument(flags=("path",), dest="path", positional=True, choices=["a", "b"])
assert arg.get_positional_text() == "{a,b}"
def test_positional_text_without_choices():
arg = Argument(flags=("path",), dest="path", positional=True)
assert arg.get_positional_text() == "path"
@pytest.mark.parametrize(
"nargs,expected",
[
(None, "VALUE"),
(1, "VALUE"),
("?", "[VALUE]"),
("*", "[VALUE ...]"),
("+", "VALUE [VALUE ...]"),
],
)
def test_choice_text_store_action_variants(nargs, expected):
arg = Argument(
flags=("--value",), dest="value", action=ArgumentAction.STORE, nargs=nargs
)
assert arg.get_choice_text() == expected
@pytest.mark.parametrize(
"nargs,expected",
[
(None, "value"),
(1, "value"),
("?", "[value]"),
("*", "[value ...]"),
("+", "value [value ...]"),
],
)
def test_choice_text_store_action_variants_positional(nargs, expected):
arg = Argument(
flags=("value",),
dest="value",
action=ArgumentAction.STORE,
nargs=nargs,
positional=True,
)
assert arg.get_choice_text() == expected
def test_choice_text_with_choices():
arg = Argument(flags=("--mode",), dest="mode", choices=["dev", "prod"])
assert arg.get_choice_text() == "{dev,prod}"
def test_choice_text_append_and_extend():
for action in [ArgumentAction.APPEND, ArgumentAction.EXTEND]:
arg = Argument(flags=("--tag",), dest="tag", action=action)
assert arg.get_choice_text() == "TAG"
def test_equality():
a1 = Argument(flags=("--f",), dest="f")
a2 = Argument(flags=("--f",), dest="f")
a3 = Argument(flags=("-x",), dest="x")
assert a1 == a2
assert a1 != a3
assert hash(a1) == hash(a2)
def test_inequality_with_non_argument():
arg = Argument(flags=("--f",), dest="f")
assert arg != "not an argument"
def test_argument_equality():
arg = Argument("--foo", dest="foo", type=str, default="default_value")
arg2 = Argument("--foo", dest="foo", type=str, default="default_value")
arg3 = Argument("--bar", dest="bar", type=int, default=42)
arg4 = Argument("--foo", dest="foo", type=str, default="foobar")
assert arg == arg2
assert arg != arg3
assert arg != arg4
assert arg != "not an argument"
assert arg is not None
assert arg != object()

11
tests/test_parsers/test_argument_action.py Normal file
View File

@ -0,0 +1,11 @@
from falyx.parsers import ArgumentAction
def test_argument_action():
action = ArgumentAction.APPEND
assert action == ArgumentAction.APPEND
assert action != ArgumentAction.STORE
assert action != "invalid_action"
assert action.value == "append"
assert str(action) == "append"
assert len(ArgumentAction.choices()) == 8

0
tests/test_parsers/test_basics.py Normal file
View File

56
tests/test_parsers/test_nargs.py Normal file
View File

@ -0,0 +1,56 @@
import pytest
from falyx.exceptions import CommandArgumentError
from falyx.parsers import ArgumentAction, CommandArgumentParser
@pytest.mark.asyncio
async def test_nargs():
"""Test the nargs argument for command-line arguments."""
parser = CommandArgumentParser()
parser.add_argument(
"-a",
"--alpha",
action=ArgumentAction.STORE,
nargs=2,
help="Alpha option with two arguments",
)
parser.add_argument(
"-b",
"--beta",
action=ArgumentAction.STORE,
nargs="+",
help="Beta option with one or more arguments",
)
parser.add_argument(
"-c",
"--charlie",
action=ArgumentAction.STORE,
nargs="*",
help="Charlie option with zero or more arguments",
)
# Test valid cases
args = await parser.parse_args(["-a", "value1", "value2"])
assert args["alpha"] == ["value1", "value2"]
args = await parser.parse_args(["-b", "value1", "value2", "value3"])
assert args["beta"] == ["value1", "value2", "value3"]
args = await parser.parse_args(["-c", "value1", "value2"])
assert args["charlie"] == ["value1", "value2"]
args = await parser.parse_args(["-c"])
assert args["charlie"] == []
with pytest.raises(CommandArgumentError):
await parser.parse_args(["-a", "value1"])
with pytest.raises(CommandArgumentError):
await parser.parse_args(["-a"])
with pytest.raises(CommandArgumentError):
await parser.parse_args(["-a", "value1", "value2", "value3"])
with pytest.raises(CommandArgumentError):
await parser.parse_args(["-b"])

128
tests/test_parsers/test_posix_bundling.py Normal file
View File

@ -0,0 +1,128 @@
import pytest
from falyx.exceptions import CommandArgumentError
from falyx.parsers import ArgumentAction, CommandArgumentParser
@pytest.mark.asyncio
async def test_posix_bundling():
"""Test the bundling of short options in the POSIX style."""
parser = CommandArgumentParser()
parser.add_argument(
"-a", "--alpha", action=ArgumentAction.STORE_FALSE, help="Alpha option"
)
parser.add_argument(
"-b", "--beta", action=ArgumentAction.STORE_TRUE, help="Beta option"
)
parser.add_argument(
"-c", "--charlie", action=ArgumentAction.STORE_TRUE, help="Charlie option"
)
# Test valid bundling
args = await parser.parse_args(["-abc"])
assert args["alpha"] is False
assert args["beta"] is True
assert args["charlie"] is True
@pytest.mark.asyncio
async def test_posix_bundling_last_has_value():
"""Test the bundling of short options in the POSIX style with last option having a value."""
parser = CommandArgumentParser()
parser.add_argument(
"-a", "--alpha", action=ArgumentAction.STORE_TRUE, help="Alpha option"
)
parser.add_argument(
"-b", "--beta", action=ArgumentAction.STORE_TRUE, help="Beta option"
)
parser.add_argument(
"-c", "--charlie", action=ArgumentAction.STORE, help="Charlie option"
)
# Test valid bundling with last option having a value
args = await parser.parse_args(["-abc", "value"])
assert args["alpha"] is True
assert args["beta"] is True
assert args["charlie"] == "value"
@pytest.mark.asyncio
async def test_posix_bundling_invalid():
"""Test the bundling of short options in the POSIX style with invalid cases."""
parser = CommandArgumentParser()
parser.add_argument(
"-a", "--alpha", action=ArgumentAction.STORE_FALSE, help="Alpha option"
)
parser.add_argument(
"-b", "--beta", action=ArgumentAction.STORE_TRUE, help="Beta option"
)
parser.add_argument(
"-c", "--charlie", action=ArgumentAction.STORE, help="Charlie option"
)
# Test invalid bundling
args = await parser.parse_args(["-abc", "value"])
assert args["alpha"] is False
assert args["beta"] is True
assert args["charlie"] == "value"
with pytest.raises(CommandArgumentError):
await parser.parse_args(["-a", "value"])
with pytest.raises(CommandArgumentError):
await parser.parse_args(["-b", "value"])
args = await parser.parse_args(["-c", "value"])
assert args["alpha"] is True
assert args["beta"] is False
assert args["charlie"] == "value"
with pytest.raises(CommandArgumentError):
await parser.parse_args(["-cab", "value"])
with pytest.raises(CommandArgumentError):
await parser.parse_args(["-a", "-b", "value"])
with pytest.raises(CommandArgumentError):
await parser.parse_args(["-dbc", "value"])
with pytest.raises(CommandArgumentError):
args = await parser.parse_args(["-c"])
with pytest.raises(CommandArgumentError):
await parser.parse_args(["-abc"])
@pytest.mark.asyncio
async def test_posix_bundling_fuzz():
"""Test the bundling of short options in the POSIX style with fuzzing."""
parser = CommandArgumentParser()
parser.add_argument(
"-a", "--alpha", action=ArgumentAction.STORE_FALSE, help="Alpha option"
)
with pytest.raises(CommandArgumentError):
await parser.parse_args(["-"])
with pytest.raises(CommandArgumentError):
await parser.parse_args(["--"])
with pytest.raises(CommandArgumentError):
await parser.parse_args(["--=value"])
with pytest.raises(CommandArgumentError):
await parser.parse_args(["--flag="])
with pytest.raises(CommandArgumentError):
await parser.parse_args(["-a=b"])
with pytest.raises(CommandArgumentError):
await parser.parse_args(["---"])
with pytest.raises(CommandArgumentError):
await parser.parse_args(["-a", "-b", "-c"])
with pytest.raises(CommandArgumentError):
await parser.parse_args(["-a", "--", "-b", "-c"])
with pytest.raises(CommandArgumentError):
await parser.parse_args(["-a", "--flag", "-b", "-c"])

3
tests/test_run_key.py
View File

@ -1,6 +1,7 @@
import pytest
from falyx import Action, Falyx
from falyx import Falyx
from falyx.action import Action
@pytest.mark.asyncio