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

Move ShellAction to action/shell_action.py, Move Argument, ArgumentAction, and CommandArgumentParser to seperate files

Browse Source
This commit is contained in:
Roland Thomas Jr
2025-06-08 14:31:24 -04:00
parent 53ba6a896a
commit 3c7ef3eb1c
9 changed files with 233 additions and 212 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
falyx/action/__init__.py
View File

@ -12,7 +12,7 @@ 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 .io_action import BaseIOAction
from .literal_input_action import LiteralInputAction
from .menu_action import MenuAction
from .process_action import ProcessAction
@ -20,6 +20,7 @@ from .process_pool_action import ProcessPoolAction
from .prompt_menu_action import PromptMenuAction
from .select_file_action import SelectFileAction
from .selection_action import SelectionAction
from .shell_action import ShellAction
from .signal_action import SignalAction
from .user_input_action import UserInputAction

93
falyx/action/io_action.py
View File

@ -16,8 +16,6 @@ Common usage includes shell-like filters, input transformers, or any tool that
needs to consume input from another process or pipeline.
"""
import asyncio
import shlex
import subprocess
import sys
from typing import Any, Callable
@ -25,10 +23,8 @@ from rich.tree import Tree
from falyx.action.base import BaseAction
from falyx.context import ExecutionContext
from falyx.exceptions import FalyxError
from falyx.execution_registry import ExecutionRegistry as er
from falyx.hook_manager import HookManager, HookType
from falyx.logger import logger
from falyx.themes import OneColors
@ -171,92 +167,3 @@ class BaseIOAction(BaseAction):
parent.add("".join(label))
else:
self.console.print(Tree("".join(label)))
class ShellAction(BaseIOAction):
"""
ShellAction wraps a shell command template for CLI pipelines.
This Action takes parsed input (from stdin, literal, or last_result),
substitutes it into the provided shell command template, and executes
the command asynchronously using subprocess.
Designed for quick integration with shell tools like `grep`, `ping`, `jq`, etc.
⚠️ Security Warning:
By default, ShellAction uses `shell=True`, which can be dangerous with
unsanitized input. To mitigate this, set `safe_mode=True` to use `shell=False`
with `shlex.split()`.
Features:
- Automatically handles input parsing (str/bytes)
- `safe_mode=True` disables shell interpretation and runs with `shell=False`
- Captures stdout and stderr from shell execution
- Raises on non-zero exit codes with stderr as the error
- Result is returned as trimmed stdout string
Args:
name (str): Name of the action.
command_template (str): Shell command to execute. Must include `{}` to include
input. If no placeholder is present, the input is not
included.
safe_mode (bool): If True, runs with `shell=False` using shlex parsing
(default: False).
"""
def __init__(
self, name: str, command_template: str, safe_mode: bool = False, **kwargs
):
super().__init__(name=name, **kwargs)
self.command_template = command_template
self.safe_mode = safe_mode
def from_input(self, raw: str | bytes) -> str:
if not isinstance(raw, (str, bytes)):
raise TypeError(
f"{self.name} expected str or bytes input, got {type(raw).__name__}"
)
return raw.strip() if isinstance(raw, str) else raw.decode("utf-8").strip()
def get_infer_target(self) -> tuple[Callable[..., Any] | None, dict[str, Any] | None]:
if sys.stdin.isatty():
return self._run, {"parsed_input": {"help": self.command_template}}
return None, None
async def _run(self, parsed_input: str) -> str:
# Replace placeholder in template, or use raw input as full command
command = self.command_template.format(parsed_input)
if self.safe_mode:
try:
args = shlex.split(command)
except ValueError as error:
raise FalyxError(f"Invalid command template: {error}")
result = subprocess.run(args, capture_output=True, text=True, check=True)
else:
result = subprocess.run(
command, shell=True, text=True, capture_output=True, check=True
)
if result.returncode != 0:
raise RuntimeError(result.stderr.strip())
return result.stdout.strip()
def to_output(self, result: str) -> str:
return result
async def preview(self, parent: Tree | None = None):
label = [f"[{OneColors.GREEN_b}]⚙ ShellAction[/] '{self.name}'"]
label.append(f"\n[dim]Template:[/] {self.command_template}")
label.append(
f"\n[dim]Safe mode:[/] {'Enabled' if self.safe_mode else 'Disabled'}"
)
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))
if not parent:
self.console.print(tree)
def __str__(self):
return (
f"ShellAction(name={self.name!r}, command_template={self.command_template!r},"
f" safe_mode={self.safe_mode})"
)

103
falyx/action/shell_action.py Normal file
View File

@ -0,0 +1,103 @@
"""shell_action.py - Execute shell commands with input substitution."""
from __future__ import annotations
import shlex
import subprocess
import sys
from typing import Any, Callable
from rich.tree import Tree
from falyx.action.io_action import BaseIOAction
from falyx.exceptions import FalyxError
from falyx.themes import OneColors
class ShellAction(BaseIOAction):
"""
ShellAction wraps a shell command template for CLI pipelines.
This Action takes parsed input (from stdin, literal, or last_result),
substitutes it into the provided shell command template, and executes
the command asynchronously using subprocess.
Designed for quick integration with shell tools like `grep`, `ping`, `jq`, etc.
⚠️ Security Warning:
By default, ShellAction uses `shell=True`, which can be dangerous with
unsanitized input. To mitigate this, set `safe_mode=True` to use `shell=False`
with `shlex.split()`.
Features:
- Automatically handles input parsing (str/bytes)
- `safe_mode=True` disables shell interpretation and runs with `shell=False`
- Captures stdout and stderr from shell execution
- Raises on non-zero exit codes with stderr as the error
- Result is returned as trimmed stdout string
Args:
name (str): Name of the action.
command_template (str): Shell command to execute. Must include `{}` to include
input. If no placeholder is present, the input is not
included.
safe_mode (bool): If True, runs with `shell=False` using shlex parsing
(default: False).
"""
def __init__(
self, name: str, command_template: str, safe_mode: bool = False, **kwargs
):
super().__init__(name=name, **kwargs)
self.command_template = command_template
self.safe_mode = safe_mode
def from_input(self, raw: str | bytes) -> str:
if not isinstance(raw, (str, bytes)):
raise TypeError(
f"{self.name} expected str or bytes input, got {type(raw).__name__}"
)
return raw.strip() if isinstance(raw, str) else raw.decode("utf-8").strip()
def get_infer_target(self) -> tuple[Callable[..., Any] | None, dict[str, Any] | None]:
if sys.stdin.isatty():
return self._run, {"parsed_input": {"help": self.command_template}}
return None, None
async def _run(self, parsed_input: str) -> str:
# Replace placeholder in template, or use raw input as full command
command = self.command_template.format(parsed_input)
if self.safe_mode:
try:
args = shlex.split(command)
except ValueError as error:
raise FalyxError(f"Invalid command template: {error}")
result = subprocess.run(args, capture_output=True, text=True, check=True)
else:
result = subprocess.run(
command, shell=True, text=True, capture_output=True, check=True
)
if result.returncode != 0:
raise RuntimeError(result.stderr.strip())
return result.stdout.strip()
def to_output(self, result: str) -> str:
return result
async def preview(self, parent: Tree | None = None):
label = [f"[{OneColors.GREEN_b}]⚙ ShellAction[/] '{self.name}'"]
label.append(f"\n[dim]Template:[/] {self.command_template}")
label.append(
f"\n[dim]Safe mode:[/] {'Enabled' if self.safe_mode else 'Disabled'}"
)
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))
if not parent:
self.console.print(tree)
def __str__(self):
return (
f"ShellAction(name={self.name!r}, command_template={self.command_template!r},"
f" safe_mode={self.safe_mode})"
)

4
falyx/parser/__init__.py
View File

@ -5,7 +5,9 @@ Copyright (c) 2025 rtj.dev LLC.
Licensed under the MIT License. See LICENSE file for details.
"""
from .argparse import Argument, ArgumentAction, CommandArgumentParser
from .argument import Argument
from .argument_action import ArgumentAction
from .command_argument_parser import CommandArgumentParser
from .parsers import FalyxParsers, get_arg_parsers, get_root_parser, get_subparsers
__all__ = [

96
falyx/parser/argument.py Normal file
View File

@ -0,0 +1,96 @@
from dataclasses import dataclass
from typing import Any
from falyx.action.base import BaseAction
from falyx.parser.argument_action import ArgumentAction
@dataclass
class Argument:
"""Represents a command-line argument."""
flags: tuple[str, ...]
dest: str # Destination name for the argument
action: ArgumentAction = (
ArgumentAction.STORE
) # Action to be taken when the argument is encountered
type: Any = str # Type of the argument (e.g., str, int, float) or callable
default: Any = None # Default value if the argument is not provided
choices: list[str] | None = None # List of valid choices for the argument
required: bool = False # True if the argument is required
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."""
text = ""
if self.positional:
if self.choices:
text = f"{{{','.join([str(choice) for choice in self.choices])}}}"
else:
text = self.dest
return text
def get_choice_text(self) -> str:
"""Get the choice text for the argument."""
choice_text = ""
if self.choices:
choice_text = f"{{{','.join([str(choice) for choice in self.choices])}}}"
elif (
self.action
in (
ArgumentAction.STORE,
ArgumentAction.APPEND,
ArgumentAction.EXTEND,
)
and not self.positional
):
choice_text = self.dest.upper()
elif self.action in (
ArgumentAction.STORE,
ArgumentAction.APPEND,
ArgumentAction.EXTEND,
) or isinstance(self.nargs, str):
choice_text = self.dest
if self.nargs == "?":
choice_text = f"[{choice_text}]"
elif self.nargs == "*":
choice_text = f"[{choice_text} ...]"
elif self.nargs == "+":
choice_text = f"{choice_text} [{choice_text} ...]"
return choice_text
def __eq__(self, other: object) -> bool:
if not isinstance(other, Argument):
return False
return (
self.flags == other.flags
and self.dest == other.dest
and self.action == other.action
and self.type == other.type
and self.choices == other.choices
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:
return hash(
(
tuple(self.flags),
self.dest,
self.action,
self.type,
tuple(self.choices or []),
self.required,
self.nargs,
self.positional,
self.default,
self.help,
)
)

25
falyx/parser/argument_action.py Normal file
View File

@ -0,0 +1,25 @@
from __future__ import annotations
from enum import Enum
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"
APPEND = "append"
EXTEND = "extend"
COUNT = "count"
HELP = "help"
@classmethod
def choices(cls) -> list[ArgumentAction]:
"""Return a list of all argument actions."""
return list(cls)
def __str__(self) -> str:
"""Return the string representation of the argument action."""
return self.value

117
falyx/parser/argparse.py → falyx/parser/command_argument_parser.py
View File

@ -2,8 +2,6 @@
from __future__ import annotations
from copy import deepcopy
from dataclasses import dataclass
from enum import Enum
from typing import Any, Iterable
from rich.console import Console
@ -12,123 +10,12 @@ from rich.text import Text
from falyx.action.base import BaseAction
from falyx.exceptions import CommandArgumentError
from falyx.parser.argument import Argument
from falyx.parser.argument_action import ArgumentAction
from falyx.parser.utils import coerce_value
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"
APPEND = "append"
EXTEND = "extend"
COUNT = "count"
HELP = "help"
@classmethod
def choices(cls) -> list[ArgumentAction]:
"""Return a list of all argument actions."""
return list(cls)
def __str__(self) -> str:
"""Return the string representation of the argument action."""
return self.value
@dataclass
class Argument:
"""Represents a command-line argument."""
flags: tuple[str, ...]
dest: str # Destination name for the argument
action: ArgumentAction = (
ArgumentAction.STORE
) # Action to be taken when the argument is encountered
type: Any = str # Type of the argument (e.g., str, int, float) or callable
default: Any = None # Default value if the argument is not provided
choices: list[str] | None = None # List of valid choices for the argument
required: bool = False # True if the argument is required
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."""
text = ""
if self.positional:
if self.choices:
text = f"{{{','.join([str(choice) for choice in self.choices])}}}"
else:
text = self.dest
return text
def get_choice_text(self) -> str:
"""Get the choice text for the argument."""
choice_text = ""
if self.choices:
choice_text = f"{{{','.join([str(choice) for choice in self.choices])}}}"
elif (
self.action
in (
ArgumentAction.STORE,
ArgumentAction.APPEND,
ArgumentAction.EXTEND,
)
and not self.positional
):
choice_text = self.dest.upper()
elif self.action in (
ArgumentAction.STORE,
ArgumentAction.APPEND,
ArgumentAction.EXTEND,
) or isinstance(self.nargs, str):
choice_text = self.dest
if self.nargs == "?":
choice_text = f"[{choice_text}]"
elif self.nargs == "*":
choice_text = f"[{choice_text} ...]"
elif self.nargs == "+":
choice_text = f"{choice_text} [{choice_text} ...]"
return choice_text
def __eq__(self, other: object) -> bool:
if not isinstance(other, Argument):
return False
return (
self.flags == other.flags
and self.dest == other.dest
and self.action == other.action
and self.type == other.type
and self.choices == other.choices
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:
return hash(
(
tuple(self.flags),
self.dest,
self.action,
self.type,
tuple(self.choices or []),
self.required,
self.nargs,
self.positional,
self.default,
self.help,
)
)
class CommandArgumentParser:
"""
Custom argument parser for Falyx Commands.

2
falyx/version.py
View File

@ -1 +1 @@
__version__ = "0.1.51"
__version__ = "0.1.52"

2
pyproject.toml
View File

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