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

Add ActionFactoryAction, Add mode flags for Falyx, Rename inject_last_result_as -> inject_into

This commit is contained in:
Roland Thomas Jr 2025-05-09 23:43:36 -04:00
parent ad803e01be
commit 76e542cfce
Signed by: roland
GPG Key ID: 7C3C2B085A4C2872
10 changed files with 179 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
falyx/action.py
View File

@ -56,7 +56,7 @@ class BaseAction(ABC):
be run independently or as part of Falyx.
inject_last_result (bool): Whether to inject the previous action's result into kwargs.
inject_last_result_as (str): The name of the kwarg key to inject the result as
inject_into (str): The name of the kwarg key to inject the result as
(default: 'last_result').
_requires_injection (bool): Whether the action requires input injection.
"""
@ -66,7 +66,7 @@ class BaseAction(ABC):
name: str,
hooks: HookManager | None = None,
inject_last_result: bool = False,
inject_last_result_as: str = "last_result",
inject_into: str = "last_result",
never_prompt: bool = False,
logging_hooks: bool = False,
) -> None:
@ -75,7 +75,7 @@ class BaseAction(ABC):
self.is_retryable: bool = False
self.shared_context: SharedContext | None = None
self.inject_last_result: bool = inject_last_result
self.inject_last_result_as: str = inject_last_result_as
self.inject_into: str = inject_into
self._never_prompt: bool = never_prompt
self._requires_injection: bool = False
self._skip_in_chain: bool = False
@ -133,7 +133,7 @@ class BaseAction(ABC):
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_last_result_as
key = self.inject_into
if key in kwargs:
logger.warning("[%s] ⚠️ Overriding '%s' with last_result", self.name, key)
kwargs = dict(kwargs)
@ -173,7 +173,7 @@ class Action(BaseAction):
kwargs (dict, optional): Static keyword arguments.
hooks (HookManager, optional): Hook manager for lifecycle events.
inject_last_result (bool, optional): Enable last_result injection.
inject_last_result_as (str, optional): Name of injected key.
inject_into (str, optional): Name of injected key.
retry (bool, optional): Enable retry logic.
retry_policy (RetryPolicy, optional): Retry settings.
"""
@ -187,11 +187,11 @@ class Action(BaseAction):
kwargs: dict[str, Any] | None = None,
hooks: HookManager | None = None,
inject_last_result: bool = False,
inject_last_result_as: str = "last_result",
inject_into: str = "last_result",
retry: bool = False,
retry_policy: RetryPolicy | None = None,
) -> None:
super().__init__(name, hooks, inject_last_result, inject_last_result_as)
super().__init__(name, hooks, inject_last_result, inject_into)
self.action = action
self.rollback = rollback
self.args = args
@ -257,7 +257,7 @@ class Action(BaseAction):
if context.result is not None:
logger.info("[%s] ✅ Recovered: %s", self.name, self.name)
return context.result
raise error
raise
finally:
context.stop_timer()
await self.hooks.trigger(HookType.AFTER, context)
@ -267,7 +267,7 @@ class Action(BaseAction):
async def preview(self, parent: Tree | None = None):
label = [f"[{OneColors.GREEN_b}]⚙ Action[/] '{self.name}'"]
if self.inject_last_result:
label.append(f" [dim](injects '{self.inject_last_result_as}')[/dim]")
label.append(f" [dim](injects '{self.inject_into}')[/dim]")
if self.retry_policy.enabled:
label.append(
f"\n[dim]↻ Retries:[/] {self.retry_policy.max_retries}x, "
@ -413,7 +413,7 @@ class ChainedAction(BaseAction, ActionListMixin):
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_last_result_as (str, optional): Key name for injection.
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.
"""
@ -424,11 +424,11 @@ class ChainedAction(BaseAction, ActionListMixin):
actions: list[BaseAction | Any] | None = None,
hooks: HookManager | None = None,
inject_last_result: bool = False,
inject_last_result_as: str = "last_result",
inject_into: str = "last_result",
auto_inject: bool = False,
return_list: bool = False,
) -> None:
super().__init__(name, hooks, inject_last_result, inject_last_result_as)
super().__init__(name, hooks, inject_last_result, inject_into)
ActionListMixin.__init__(self)
self.auto_inject = auto_inject
self.return_list = return_list
@ -482,9 +482,7 @@ class ChainedAction(BaseAction, ActionListMixin):
last_result = shared_context.last_result()
try:
if self.requires_io_injection() and last_result is not None:
result = await prepared(
**{prepared.inject_last_result_as: last_result}
)
result = await prepared(**{prepared.inject_into: last_result})
else:
result = await prepared(*args, **updated_kwargs)
except Exception as error:
@ -559,7 +557,7 @@ class ChainedAction(BaseAction, ActionListMixin):
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_last_result_as}')[/dim]")
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)
@ -603,7 +601,7 @@ class ActionGroup(BaseAction, ActionListMixin):
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_last_result_as (str, optional): Key name for injection.
inject_into (str, optional): Key name for injection.
"""
def __init__(
@ -612,9 +610,9 @@ class ActionGroup(BaseAction, ActionListMixin):
actions: list[BaseAction] | None = None,
hooks: HookManager | None = None,
inject_last_result: bool = False,
inject_last_result_as: str = "last_result",
inject_into: str = "last_result",
):
super().__init__(name, hooks, inject_last_result, inject_last_result_as)
super().__init__(name, hooks, inject_last_result, inject_into)
ActionListMixin.__init__(self)
if actions:
self.set_actions(actions)
@ -694,7 +692,7 @@ class ActionGroup(BaseAction, ActionListMixin):
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_last_result_as}')[/dim]")
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)
@ -726,7 +724,7 @@ class ProcessAction(BaseAction):
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_last_result_as (str, optional): Name of the injected key.
inject_into (str, optional): Name of the injected key.
"""
def __init__(
@ -738,9 +736,9 @@ class ProcessAction(BaseAction):
hooks: HookManager | None = None,
executor: ProcessPoolExecutor | None = None,
inject_last_result: bool = False,
inject_last_result_as: str = "last_result",
inject_into: str = "last_result",
):
super().__init__(name, hooks, inject_last_result, inject_last_result_as)
super().__init__(name, hooks, inject_last_result, inject_into)
self.func = func
self.args = args
self.kwargs = kwargs or {}
@ -800,7 +798,7 @@ class ProcessAction(BaseAction):
f"[{OneColors.DARK_YELLOW_b}]🧠 ProcessAction (new process)[/] '{self.name}'"
]
if self.inject_last_result:
label.append(f" [dim](injects '{self.inject_last_result_as}')[/dim]")
label.append(f" [dim](injects '{self.inject_into}')[/dim]")
if parent:
parent.add("".join(label))
else:

95
falyx/action_factory.py Normal file
View File

@ -0,0 +1,95 @@
from typing import Any
from rich.tree import Tree
from falyx.action import BaseAction
from falyx.context import ExecutionContext
from falyx.execution_registry import ExecutionRegistry as er
from falyx.hook_manager import HookType
from falyx.protocols import ActionFactoryProtocol
from falyx.themes.colors import OneColors
class ActionFactoryAction(BaseAction):
"""
Dynamically creates and runs another Action at runtime using a factory function.
This is useful for generating context-specific behavior (e.g., dynamic HTTPActions)
where the structure of the next action depends on runtime values.
Args:
name (str): Name of the action.
factory (Callable): A function that returns a BaseAction given args/kwargs.
inject_last_result (bool): Whether to inject last_result into the factory.
inject_into (str): The name of the kwarg to inject last_result as.
"""
def __init__(
self,
name: str,
factory: ActionFactoryProtocol,
*,
inject_last_result: bool = False,
inject_into: str = "last_result",
preview_args: tuple[Any, ...] = (),
preview_kwargs: dict[str, Any] = {},
):
super().__init__(
name=name,
inject_last_result=inject_last_result,
inject_into=inject_into,
)
self.factory = factory
self.preview_args = preview_args
self.preview_kwargs = preview_kwargs
async def _run(self, *args, **kwargs) -> Any:
updated_kwargs = self._maybe_inject_last_result(kwargs)
context = ExecutionContext(
name=f"{self.name} (factory)",
args=args,
kwargs=updated_kwargs,
action=self,
)
context.start_timer()
try:
await self.hooks.trigger(HookType.BEFORE, context)
generated_action = self.factory(*args, **updated_kwargs)
if not isinstance(generated_action, BaseAction):
raise TypeError(
f"[{self.name}] Factory must return a BaseAction, got {type(generated_action).__name__}"
)
if self.shared_context:
generated_action.set_shared_context(self.shared_context)
if self.options_manager:
generated_action.set_options_manager(self.options_manager)
context.result = await generated_action(*args, **kwargs)
await self.hooks.trigger(HookType.ON_SUCCESS, context)
return context.result
except Exception as error:
context.exception = error
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 preview(self, parent: Tree | None = None):
label = f"[{OneColors.CYAN_b}]🏗️ ActionFactory[/] '{self.name}'"
tree = parent.add(label) if parent else Tree(label)
try:
generated = self.factory(*self.preview_args, **self.preview_kwargs)
if isinstance(generated, BaseAction):
await generated.preview(parent=tree)
else:
tree.add(
f"[{OneColors.DARK_RED}]⚠️ Factory did not return a BaseAction[/]"
)
except Exception as error:
tree.add(f"[{OneColors.DARK_RED}]⚠️ Preview failed: {error}[/]")
if not parent:
self.console.print(tree)

49
falyx/falyx.py
View File

@ -24,6 +24,7 @@ import logging
import sys
from argparse import Namespace
from difflib import get_close_matches
from enum import Enum
from functools import cached_property
from typing import Any, Callable
@ -59,6 +60,13 @@ from falyx.utils import CaseInsensitiveDict, chunks, get_program_invocation, log
from falyx.version import __version__
class FalyxMode(str, Enum):
MENU = "menu"
RUN = "run"
PREVIEW = "preview"
RUN_ALL = "run-all"
class Falyx:
"""
Main menu controller for Falyx CLI applications.
@ -149,6 +157,7 @@ class Falyx:
self.custom_table: Callable[["Falyx"], Table] | Table | None = custom_table
self.validate_options(cli_args, options)
self._prompt_session: PromptSession | None = None
self.mode = FalyxMode.MENU
def validate_options(
self,
@ -272,6 +281,11 @@ class Falyx:
)
self.console.print(table, justify="center")
if self.mode == FalyxMode.MENU:
self.console.print(
f"📦 Tip: Type '[{OneColors.LIGHT_YELLOW}]?[KEY][/]' to preview a command before running it.\n",
justify="center",
)
def _get_help_command(self) -> Command:
"""Returns the help command for the menu."""
@ -329,7 +343,8 @@ class Falyx:
error_message = " ".join(message_lines)
def validator(text):
return True if self.get_command(text, from_validate=True) else False
_, choice = self.get_command(text, from_validate=True)
return True if choice else False
return Validator.from_callable(
validator,
@ -668,17 +683,25 @@ class Falyx:
else:
return self.build_default_table()
def get_command(self, choice: str, from_validate=False) -> Command | None:
def parse_preview_command(self, input_str: str) -> tuple[bool, str]:
if input_str.startswith("?"):
return True, input_str[1:].strip()
return False, input_str.strip()
def get_command(
self, choice: str, from_validate=False
) -> tuple[bool, Command | None]:
"""Returns the selected command based on user input. Supports keys, aliases, and abbreviations."""
is_preview, choice = self.parse_preview_command(choice)
choice = choice.upper()
name_map = self._name_map
if choice in name_map:
return name_map[choice]
return is_preview, name_map[choice]
prefix_matches = [cmd for key, cmd in name_map.items() if key.startswith(choice)]
if len(prefix_matches) == 1:
return prefix_matches[0]
return is_preview, prefix_matches[0]
fuzzy_matches = get_close_matches(choice, list(name_map.keys()), n=3, cutoff=0.7)
if fuzzy_matches:
@ -694,7 +717,7 @@ class Falyx:
self.console.print(
f"[{OneColors.LIGHT_YELLOW}]⚠️ Unknown command '{choice}'[/]"
)
return None
return is_preview, None
def _create_context(self, selected_command: Command) -> ExecutionContext:
"""Creates a context dictionary for the selected command."""
@ -718,11 +741,16 @@ class Falyx:
async def process_command(self) -> bool:
"""Processes the action of the selected command."""
choice = await self.prompt_session.prompt_async()
selected_command = self.get_command(choice)
is_preview, selected_command = self.get_command(choice)
if not selected_command:
logger.info(f"Invalid command '{choice}'.")
return True
if is_preview:
logger.info(f"Preview command '{selected_command.key}' selected.")
await selected_command.preview()
return True
if selected_command.requires_input:
program = get_program_invocation()
self.console.print(
@ -759,7 +787,7 @@ class Falyx:
async def run_key(self, command_key: str, return_context: bool = False) -> Any:
"""Run a command by key without displaying the menu (non-interactive mode)."""
self.debug_hooks()
selected_command = self.get_command(command_key)
_, selected_command = self.get_command(command_key)
self.last_run_command = selected_command
if not selected_command:
@ -899,7 +927,8 @@ class Falyx:
sys.exit(0)
if self.cli_args.command == "preview":
command = self.get_command(self.cli_args.name)
self.mode = FalyxMode.PREVIEW
_, command = self.get_command(self.cli_args.name)
if not command:
self.console.print(
f"[{OneColors.DARK_RED}]❌ Command '{self.cli_args.name}' not found.[/]"
@ -912,7 +941,8 @@ class Falyx:
sys.exit(0)
if self.cli_args.command == "run":
command = self.get_command(self.cli_args.name)
self.mode = FalyxMode.RUN
_, command = self.get_command(self.cli_args.name)
if not command:
self.console.print(
f"[{OneColors.DARK_RED}]❌ Command '{self.cli_args.name}' not found.[/]"
@ -927,6 +957,7 @@ class Falyx:
sys.exit(0)
if self.cli_args.command == "run-all":
self.mode = FalyxMode.RUN_ALL
matching = [
cmd
for cmd in self.commands.values()

8
falyx/http_action.py
View File

@ -56,7 +56,7 @@ class HTTPAction(Action):
data (Any, optional): Raw data or form-encoded body.
hooks (HookManager, optional): Hook manager for lifecycle events.
inject_last_result (bool): Enable last_result injection.
inject_last_result_as (str): Name of injected key.
inject_into (str): Name of injected key.
retry (bool): Enable retry logic.
retry_policy (RetryPolicy): Retry settings.
"""
@ -74,7 +74,7 @@ class HTTPAction(Action):
data: Any = None,
hooks=None,
inject_last_result: bool = False,
inject_last_result_as: str = "last_result",
inject_into: str = "last_result",
retry: bool = False,
retry_policy=None,
):
@ -92,7 +92,7 @@ class HTTPAction(Action):
kwargs={},
hooks=hooks,
inject_last_result=inject_last_result,
inject_last_result_as=inject_last_result_as,
inject_into=inject_into,
retry=retry,
retry_policy=retry_policy,
)
@ -138,7 +138,7 @@ class HTTPAction(Action):
f"\n[dim]URL:[/] {self.url}",
]
if self.inject_last_result:
label.append(f"\n[dim]Injects:[/] '{self.inject_last_result_as}'")
label.append(f"\n[dim]Injects:[/] '{self.inject_into}'")
if self.retry_policy and self.retry_policy.enabled:
label.append(
f"\n[dim]↻ Retries:[/] {self.retry_policy.max_retries}x, "

6
falyx/io_action.py
View File

@ -83,7 +83,7 @@ class BaseIOAction(BaseAction):
raise NotImplementedError
async def _resolve_input(self, kwargs: dict[str, Any]) -> str | bytes:
last_result = kwargs.pop(self.inject_last_result_as, None)
last_result = kwargs.pop(self.inject_into, None)
data = await self._read_stdin()
if data:
@ -168,7 +168,7 @@ class BaseIOAction(BaseAction):
async def preview(self, parent: Tree | None = None):
label = [f"[{OneColors.GREEN_b}]⚙ IOAction[/] '{self.name}'"]
if self.inject_last_result:
label.append(f" [dim](injects '{self.inject_last_result_as}')[/dim]")
label.append(f" [dim](injects '{self.inject_into}')[/dim]")
if parent:
parent.add("".join(label))
else:
@ -243,7 +243,7 @@ class ShellAction(BaseIOAction):
async def preview(self, parent: Tree | None = None):
label = [f"[{OneColors.GREEN_b}]⚙ ShellAction[/] '{self.name}'"]
if self.inject_last_result:
label.append(f" [dim](injects '{self.inject_last_result_as}')[/dim]")
label.append(f" [dim](injects '{self.inject_into}')[/dim]")
if parent:
parent.add("".join(label))
else:

4
falyx/menu_action.py
View File

@ -101,7 +101,7 @@ class MenuAction(BaseAction):
prompt_message: str = "Select > ",
default_selection: str = "",
inject_last_result: bool = False,
inject_last_result_as: str = "last_result",
inject_into: str = "last_result",
console: Console | None = None,
prompt_session: PromptSession | None = None,
never_prompt: bool = False,
@ -111,7 +111,7 @@ class MenuAction(BaseAction):
super().__init__(
name,
inject_last_result=inject_last_result,
inject_last_result_as=inject_last_result_as,
inject_into=inject_into,
never_prompt=never_prompt,
)
self.menu_options = menu_options

9
falyx/protocols.py Normal file
View File

@ -0,0 +1,9 @@
from __future__ import annotations
from typing import Any, Protocol
from falyx.action import BaseAction
class ActionFactoryProtocol(Protocol):
def __call__(self, *args: Any, **kwargs: Any) -> BaseAction: ...

4
falyx/selection_action.py
View File

@ -33,7 +33,7 @@ class SelectionAction(BaseAction):
prompt_message: str = "Select > ",
default_selection: str = "",
inject_last_result: bool = False,
inject_last_result_as: str = "last_result",
inject_into: str = "last_result",
return_key: bool = False,
console: Console | None = None,
prompt_session: PromptSession | None = None,
@ -43,7 +43,7 @@ class SelectionAction(BaseAction):
super().__init__(
name,
inject_last_result=inject_last_result,
inject_last_result_as=inject_last_result_as,
inject_into=inject_into,
never_prompt=never_prompt,
)
self.selections: list[str] | CaseInsensitiveDict = selections

2
falyx/version.py
View File

@ -1 +1 @@
__version__ = "0.1.19"
__version__ = "0.1.20"

2
pyproject.toml
View File

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