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

Move spinner and confirmation logic from Falyx to Command

This commit is contained in:
Roland Thomas Jr 2025-05-08 00:45:24 -04:00
parent 05a7f982f2
commit 880d86d47d
Signed by: roland
GPG Key ID: 7C3C2B085A4C2872
9 changed files with 89 additions and 95 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
falyx/command.py
View File

@ -29,14 +29,16 @@ from rich.tree import Tree
from falyx.action import Action, ActionGroup, BaseAction, ChainedAction
from falyx.context import ExecutionContext
from falyx.debug import register_debug_hooks
from falyx.exceptions import FalyxError
from falyx.execution_registry import ExecutionRegistry as er
from falyx.hook_manager import HookManager, HookType
from falyx.io_action import BaseIOAction
from falyx.options_manager import OptionsManager
from falyx.prompt_utils import should_prompt_user
from falyx.retry import RetryPolicy
from falyx.retry_utils import enable_retries_recursively
from falyx.themes.colors import OneColors
from falyx.utils import _noop, ensure_async, logger
from falyx.utils import _noop, confirm_async, ensure_async, logger
console = Console()
@ -180,7 +182,10 @@ class Command(BaseModel):
self.action.set_options_manager(self.options_manager)
async def __call__(self, *args, **kwargs) -> Any:
"""Run the action with full hook lifecycle, timing, and error handling."""
"""
Run the action with full hook lifecycle, timing, error handling,
confirmation prompts, preview, and spinner integration.
"""
self._inject_options_manager()
combined_args = args + self.args
combined_kwargs = {**self.kwargs, **kwargs}
@ -191,11 +196,29 @@ class Command(BaseModel):
action=self,
)
self._context = context
if should_prompt_user(confirm=self.confirm, options=self.options_manager):
if self.preview_before_confirm:
await self.preview()
if not await confirm_async(self.confirmation_prompt):
logger.info(f"[Command:{self.key}] ❌ Cancelled by user.")
raise FalyxError(f"[Command:{self.key}] Cancelled by confirmation.")
context.start_timer()
try:
await self.hooks.trigger(HookType.BEFORE, context)
if self.spinner:
with console.status(
self.spinner_message,
spinner=self.spinner_type,
spinner_style=self.spinner_style,
**self.spinner_kwargs,
):
result = await self.action(*combined_args, **combined_kwargs)
else:
result = await self.action(*combined_args, **combined_kwargs)
context.result = result
await self.hooks.trigger(HookType.ON_SUCCESS, context)
return context.result

2
falyx/context.py
View File

@ -129,7 +129,7 @@ class ExecutionContext(BaseModel):
if self.start_wall:
message.append(f"Start: {self.start_wall.strftime('%H:%M:%S')} | ")
if self.end_time:
if self.end_wall:
message.append(f"End: {self.end_wall.strftime('%H:%M:%S')} | ")
message.append(f"Duration: {summary['duration']:.3f}s | ")

88
falyx/falyx.py
View File

@ -55,13 +55,7 @@ from falyx.parsers import get_arg_parsers
from falyx.retry import RetryPolicy
from falyx.signals import BackSignal, QuitSignal
from falyx.themes.colors import OneColors, get_nord_theme
from falyx.utils import (
CaseInsensitiveDict,
chunks,
confirm_async,
get_program_invocation,
logger,
)
from falyx.utils import CaseInsensitiveDict, chunks, get_program_invocation, logger
from falyx.version import __version__
@ -93,9 +87,8 @@ class Falyx:
key_bindings (KeyBindings | None): Custom Prompt Toolkit key bindings.
include_history_command (bool): Whether to add a built-in history viewer command.
include_help_command (bool): Whether to add a built-in help viewer command.
confirm_on_error (bool): Whether to prompt the user after errors.
never_prompt (bool): Whether to skip confirmation prompts entirely.
always_confirm (bool): Whether to force confirmation prompts for all actions.
never_prompt (bool): Seed default for `OptionsManager["never_prompt"]`
force_confirm (bool): Seed default for `OptionsManager["force_confirm"]`
cli_args (Namespace | None): Parsed CLI arguments, usually from argparse.
options (OptionsManager | None): Declarative option mappings.
custom_table (Callable[[Falyx], Table] | Table | None): Custom menu table generator.
@ -123,9 +116,8 @@ class Falyx:
key_bindings: KeyBindings | None = None,
include_history_command: bool = True,
include_help_command: bool = True,
confirm_on_error: bool = True,
never_prompt: bool = False,
always_confirm: bool = False,
force_confirm: bool = False,
cli_args: Namespace | None = None,
options: OptionsManager | None = None,
render_menu: Callable[["Falyx"], None] | None = None,
@ -150,16 +142,15 @@ class Falyx:
self.last_run_command: Command | None = None
self.key_bindings: KeyBindings = key_bindings or KeyBindings()
self.bottom_bar: BottomBar | str | Callable[[], None] = bottom_bar
self.confirm_on_error: bool = confirm_on_error
self._never_prompt: bool = never_prompt
self._always_confirm: bool = always_confirm
self._force_confirm: bool = force_confirm
self.cli_args: Namespace | None = cli_args
self.render_menu: Callable[["Falyx"], None] | None = render_menu
self.custom_table: Callable[["Falyx"], Table] | Table | None = custom_table
self.set_options(cli_args, options)
self.validate_options(cli_args, options)
self._session: PromptSession | None = None
def set_options(
def validate_options(
self,
cli_args: Namespace | None,
options: OptionsManager | None = None,
@ -175,8 +166,6 @@ class Falyx:
assert isinstance(
cli_args, Namespace
), "CLI arguments must be a Namespace object."
if options is None:
self.options.from_namespace(cli_args, "cli_args")
if not isinstance(self.options, OptionsManager):
raise FalyxError("Options must be an instance of OptionsManager.")
@ -705,33 +694,8 @@ class Falyx:
self.console.print(
f"[{OneColors.LIGHT_YELLOW}]⚠️ Unknown command '{choice}'[/]"
)
logger.warning(f"⚠️ Command '{choice}' not found.")
return None
async def _should_run_action(self, selected_command: Command) -> bool:
if self._never_prompt:
return True
if self.cli_args and getattr(self.cli_args, "skip_confirm", False):
return True
if (
self._always_confirm
or selected_command.confirm
or self.cli_args
and getattr(self.cli_args, "force_confirm", False)
):
if selected_command.preview_before_confirm:
await selected_command.preview()
confirm_answer = await confirm_async(selected_command.confirmation_prompt)
if confirm_answer:
logger.info(f"[{selected_command.description}]🔐 confirmed.")
else:
logger.info(f"[{selected_command.description}]❌ cancelled.")
return confirm_answer
return True
def _create_context(self, selected_command: Command) -> ExecutionContext:
"""Creates a context dictionary for the selected command."""
return ExecutionContext(
@ -741,16 +705,6 @@ class Falyx:
action=selected_command,
)
async def _run_action_with_spinner(self, command: Command) -> Any:
"""Runs the action of the selected command with a spinner."""
with self.console.status(
command.spinner_message,
spinner=command.spinner_type,
spinner_style=command.spinner_style,
**command.spinner_kwargs,
):
return await command()
async def _handle_action_error(
self, selected_command: Command, error: Exception
) -> None:
@ -784,18 +738,11 @@ class Falyx:
logger.info(f"🔙 Back selected: exiting {self.get_title()}")
return False
if not await self._should_run_action(selected_command):
logger.info(f"{selected_command.description} cancelled.")
return True
context = self._create_context(selected_command)
context.start_timer()
try:
await self.hooks.trigger(HookType.BEFORE, context)
if selected_command.spinner:
result = await self._run_action_with_spinner(selected_command)
else:
result = await selected_command()
context.result = result
await self.hooks.trigger(HookType.ON_SUCCESS, context)
@ -824,21 +771,10 @@ class Falyx:
selected_command.description,
)
if not await self._should_run_action(selected_command):
logger.info("[run_key] ❌ Cancelled: %s", selected_command.description)
raise FalyxError(
f"[run_key] '{selected_command.description}' "
"cancelled by confirmation."
)
context = self._create_context(selected_command)
context.start_timer()
try:
await self.hooks.trigger(HookType.BEFORE, context)
if selected_command.spinner:
result = await self._run_action_with_spinner(selected_command)
else:
result = await selected_command()
context.result = result
@ -939,6 +875,13 @@ class Falyx:
"""Run Falyx CLI with structured subcommands."""
if not self.cli_args:
self.cli_args = get_arg_parsers().root.parse_args()
self.options.from_namespace(self.cli_args, "cli_args")
if not self.options.get("never_prompt"):
self.options.set("never_prompt", self._never_prompt)
if not self.options.get("force_confirm"):
self.options.set("force_confirm", self._force_confirm)
if self.cli_args.verbose:
logging.getLogger("falyx").setLevel(logging.DEBUG)
@ -947,9 +890,6 @@ class Falyx:
logger.debug("✅ Enabling global debug hooks for all commands")
self.register_all_with_debug_hooks()
if self.cli_args.never_prompt:
self._never_prompt = True
if self.cli_args.command == "list":
await self._show_help()
sys.exit(0)

10
falyx/http_action.py
View File

@ -97,15 +97,18 @@ class HTTPAction(Action):
)
async def _request(self, *args, **kwargs) -> dict[str, Any]:
assert self.shared_context is not None, "SharedContext is not set"
# TODO: Add check for HOOK registration
if self.shared_context:
context: SharedContext = self.shared_context
session = context.get("http_session")
if session is None:
session = aiohttp.ClientSession()
context.set("http_session", session)
context.set("_session_should_close", True)
else:
session = aiohttp.ClientSession()
async with session.request(
self.method,
self.url,
@ -122,6 +125,9 @@ class HTTPAction(Action):
"body": body,
}
if not self.shared_context:
await session.close()
async def preview(self, parent: Tree | None = None):
label = [
f"[{OneColors.CYAN_b}]🌐 HTTPAction[/] '{self.name}'",

19
falyx/prompt_utils.py Normal file
View File

@ -0,0 +1,19 @@
from falyx.options_manager import OptionsManager
def should_prompt_user(
*,
confirm: bool,
options: OptionsManager,
namespace: str = "cli_args",
):
"""Determine whether to prompt the user for confirmation based on command and global options."""
never_prompt = options.get("never_prompt", False, namespace)
always_confirm = options.get("always_confirm", False, namespace)
force_confirm = options.get("force_confirm", False, namespace)
skip_confirm = options.get("skip_confirm", False, namespace)
if never_prompt or skip_confirm:
return False
return confirm or always_confirm or force_confirm

2
falyx/selection.py
View File

@ -145,7 +145,7 @@ def render_selection_indexed_table(
chunks(range(len(selections)), columns), chunks(selections, columns)
):
row = [
formatter(index, selection) if formatter else f"{index}: {selection}"
formatter(index, selection) if formatter else f"[{index}] {selection}"
for index, selection in zip(indexes, chunk)
]
table.add_row(*row)

6
falyx/selection_action.py
View File

@ -87,6 +87,12 @@ class SelectionAction(BaseAction):
if isinstance(self.selections, dict):
if maybe_result in self.selections:
effective_default = maybe_result
elif self.inject_last_result:
logger.warning(
"[%s] Injected last result '%s' not found in selections",
self.name,
maybe_result,
)
elif isinstance(self.selections, list):
if maybe_result.isdigit() and int(maybe_result) in range(
len(self.selections)

2
falyx/version.py
View File

@ -1 +1 @@
__version__ = "0.1.16"
__version__ = "0.1.17"

2
pyproject.toml
View File

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