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

Add ArgumentAction.STORE_BOOL_OPTIONAL, Add BreakChainSignal

Browse Source
This commit is contained in:
Roland Thomas Jr
2025-07-14 21:59:12 -04:00
parent 9654b9926c
commit 68d7d89d64
15 changed files with 348 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

71
examples/argument_examples.py Normal file
View File

@ -0,0 +1,71 @@
import asyncio
from enum import Enum
from falyx import Falyx
from falyx.action import Action
from falyx.parser.command_argument_parser import CommandArgumentParser
class Place(Enum):
"""Enum for different places."""
NEW_YORK = "New York"
SAN_FRANCISCO = "San Francisco"
LONDON = "London"
async def test_args(
service: str,
place: Place = Place.NEW_YORK,
region: str = "us-east-1",
verbose: bool | None = None,
) -> str:
if verbose:
print(f"Deploying {service} to {region} at {place}...")
return f"{service} deployed to {region} at {place}"
def default_config(parser: CommandArgumentParser) -> None:
"""Default argument configuration for the command."""
parser.add_argument(
"service",
type=str,
choices=["web", "database", "cache"],
help="Service name to deploy.",
)
parser.add_argument(
"place",
type=Place,
choices=list(Place),
default=Place.NEW_YORK,
help="Place where the service will be deployed.",
)
parser.add_argument(
"--region",
type=str,
default="us-east-1",
help="Deployment region.",
choices=["us-east-1", "us-west-2", "eu-west-1"],
)
parser.add_argument(
"--verbose",
action="store_bool_optional",
help="Enable verbose output.",
)
flx = Falyx("Argument Examples")
flx.add_command(
key="T",
aliases=["test"],
description="Test Command",
help_text="A command to test argument parsing.",
action=Action(
name="test_args",
action=test_args,
),
argument_config=default_config,
)
asyncio.run(flx.run())

2
falyx/action/action_types.py
View File

@ -1,5 +1,5 @@
# Falyx CLI Framework — (c) 2025 rtj.dev LLC — MIT Licensed
"""types.py"""
"""action_types.py"""
from __future__ import annotations
from enum import Enum

9
falyx/action/chained_action.py
View File

@ -17,6 +17,7 @@ from falyx.execution_registry import ExecutionRegistry as er
from falyx.hook_manager import Hook, HookManager, HookType
from falyx.logger import logger
from falyx.options_manager import OptionsManager
from falyx.signals import BreakChainSignal
from falyx.themes import OneColors
@ -106,7 +107,7 @@ class ChainedAction(BaseAction, ActionListMixin):
def _clear_args(self):
return (), {}
async def _run(self, *args, **kwargs) -> list[Any]:
async def _run(self, *args, **kwargs) -> Any:
if not self.actions:
raise EmptyChainError(f"[{self.name}] No actions to execute.")
@ -166,7 +167,11 @@ class ChainedAction(BaseAction, ActionListMixin):
context.result = all_results if self.return_list else all_results[-1]
await self.hooks.trigger(HookType.ON_SUCCESS, context)
return context.result
except BreakChainSignal as error:
logger.info("[%s] Chain broken: %s", self.name, error)
context.exception = error
shared_context.add_error(shared_context.current_index, error)
await self._rollback(context.extra["rollback_stack"], *args, **kwargs)
except Exception as error:
context.exception = error
shared_context.add_error(shared_context.current_index, error)

10
falyx/context.py
View File

@ -42,7 +42,7 @@ class ExecutionContext(BaseModel):
kwargs (dict): Keyword arguments passed to the action.
action (BaseAction | Callable): The action instance being executed.
result (Any | None): The result of the action, if successful.
exception (Exception | None): The exception raised, if execution failed.
exception (BaseException | None): The exception raised, if execution failed.
start_time (float | None): High-resolution performance start time.
end_time (float | None): High-resolution performance end time.
start_wall (datetime | None): Wall-clock timestamp when execution began.
@ -75,7 +75,7 @@ class ExecutionContext(BaseModel):
kwargs: dict = Field(default_factory=dict)
action: Any
result: Any | None = None
exception: Exception | None = None
exception: BaseException | None = None
start_time: float | None = None
end_time: float | None = None
@ -207,7 +207,7 @@ class SharedContext(BaseModel):
Attributes:
name (str): Identifier for the context (usually the parent action name).
results (list[Any]): Captures results from each action, in order of execution.
errors (list[tuple[int, Exception]]): Indexed list of errors from failed actions.
errors (list[tuple[int, BaseException]]): Indexed list of errors from failed actions.
current_index (int): Index of the currently executing action (used in chains).
is_parallel (bool): Whether the context is used in parallel mode (ActionGroup).
shared_result (Any | None): Optional shared value available to all actions in
@ -232,7 +232,7 @@ class SharedContext(BaseModel):
name: str
action: Any
results: list[Any] = Field(default_factory=list)
errors: list[tuple[int, Exception]] = Field(default_factory=list)
errors: list[tuple[int, BaseException]] = Field(default_factory=list)
current_index: int = -1
is_parallel: bool = False
shared_result: Any | None = None
@ -244,7 +244,7 @@ class SharedContext(BaseModel):
def add_result(self, result: Any) -> None:
self.results.append(result)
def add_error(self, index: int, error: Exception) -> None:
def add_error(self, index: int, error: BaseException) -> None:
self.errors.append((index, error))
def set_shared_result(self, result: Any) -> None:

1
falyx/parser/argument_action.py
View File

@ -12,6 +12,7 @@ class ArgumentAction(Enum):
STORE = "store"
STORE_TRUE = "store_true"
STORE_FALSE = "store_false"
STORE_BOOL_OPTIONAL = "store_bool_optional"
APPEND = "append"
EXTEND = "extend"
COUNT = "count"

176
falyx/parser/command_argument_parser.py
View File

@ -2,6 +2,7 @@
"""command_argument_parser.py"""
from __future__ import annotations
from collections import defaultdict
from copy import deepcopy
from typing import Any, Iterable
@ -13,6 +14,7 @@ from falyx.console import console
from falyx.exceptions import CommandArgumentError
from falyx.parser.argument import Argument
from falyx.parser.argument_action import ArgumentAction
from falyx.parser.parser_types import false_none, true_none
from falyx.parser.utils import coerce_value
from falyx.signals import HelpSignal
@ -33,6 +35,7 @@ class CommandArgumentParser:
- Support for positional and keyword arguments.
- Support for default values.
- Support for boolean flags.
- Support for optional boolean flags.
- Exception handling for invalid arguments.
- Render Help using Rich library.
"""
@ -111,10 +114,23 @@ class CommandArgumentParser:
return dest
def _determine_required(
self, required: bool, positional: bool, nargs: int | str | None
self,
required: bool,
positional: bool,
nargs: int | str | None,
action: ArgumentAction,
) -> bool:
"""Determine if the argument is required."""
if required:
if action in (
ArgumentAction.STORE_TRUE,
ArgumentAction.STORE_FALSE,
ArgumentAction.STORE_BOOL_OPTIONAL,
ArgumentAction.HELP,
):
raise CommandArgumentError(
f"Argument with action {action} cannot be required"
)
return True
if positional:
assert (
@ -143,6 +159,7 @@ class CommandArgumentParser:
ArgumentAction.STORE_TRUE,
ArgumentAction.COUNT,
ArgumentAction.HELP,
ArgumentAction.STORE_BOOL_OPTIONAL,
):
if nargs is not None:
raise CommandArgumentError(
@ -163,9 +180,17 @@ class CommandArgumentParser:
return nargs
def _normalize_choices(
self, choices: Iterable | None, expected_type: Any
self, choices: Iterable | None, expected_type: Any, action: ArgumentAction
) -> list[Any]:
if choices is not None:
if action in (
ArgumentAction.STORE_TRUE,
ArgumentAction.STORE_FALSE,
ArgumentAction.STORE_BOOL_OPTIONAL,
):
raise CommandArgumentError(
f"choices cannot be specified for {action} actions"
)
if isinstance(choices, dict):
raise CommandArgumentError("choices cannot be a dict")
try:
@ -239,6 +264,7 @@ class CommandArgumentParser:
if action in (
ArgumentAction.STORE_TRUE,
ArgumentAction.STORE_FALSE,
ArgumentAction.STORE_BOOL_OPTIONAL,
ArgumentAction.COUNT,
ArgumentAction.HELP,
):
@ -271,6 +297,14 @@ class CommandArgumentParser:
return []
else:
return None
elif action in (
ArgumentAction.STORE_TRUE,
ArgumentAction.STORE_FALSE,
ArgumentAction.STORE_BOOL_OPTIONAL,
):
raise CommandArgumentError(
f"Default value cannot be set for action {action}. It is a boolean flag."
)
return default
def _validate_flags(self, flags: tuple[str, ...]) -> None:
@ -289,6 +323,66 @@ class CommandArgumentParser:
f"Flag '{flag}' must be a single character or start with '--'"
)
def _register_store_bool_optional(
self,
flags: tuple[str, ...],
dest: str,
help: str,
) -> None:
if len(flags) != 1:
raise CommandArgumentError(
"store_bool_optional action can only have a single flag"
)
if not flags[0].startswith("--"):
raise CommandArgumentError(
"store_bool_optional action must use a long flag (e.g. --flag)"
)
base_flag = flags[0]
negated_flag = f"--no-{base_flag.lstrip('-')}"
argument = Argument(
flags=flags,
dest=dest,
action=ArgumentAction.STORE_BOOL_OPTIONAL,
type=true_none,
default=None,
help=help,
)
negated_argument = Argument(
flags=(negated_flag,),
dest=dest,
action=ArgumentAction.STORE_BOOL_OPTIONAL,
type=false_none,
default=None,
help=help,
)
self._register_argument(argument)
self._register_argument(negated_argument)
def _register_argument(self, argument: Argument):
for flag in argument.flags:
if (
flag in self._flag_map
and not argument.action == ArgumentAction.STORE_BOOL_OPTIONAL
):
existing = self._flag_map[flag]
raise CommandArgumentError(
f"Flag '{flag}' is already used by argument '{existing.dest}'"
)
for flag in argument.flags:
self._flag_map[flag] = argument
if not argument.positional:
self._keyword[flag] = argument
self._dest_set.add(argument.dest)
self._arguments.append(argument)
if argument.positional:
self._positional[argument.dest] = argument
else:
self._keyword_list.append(argument)
def add_argument(
self,
*flags,
@ -334,6 +428,7 @@ class CommandArgumentParser:
)
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 (
@ -344,46 +439,34 @@ class CommandArgumentParser:
self._validate_default_list_type(default, expected_type, dest)
else:
self._validate_default_type(default, expected_type, dest)
choices = self._normalize_choices(choices, expected_type)
choices = self._normalize_choices(choices, expected_type, action)
if default is not None and choices and default not in choices:
raise CommandArgumentError(
f"Default value '{default}' not in allowed choices: {choices}"
)
required = self._determine_required(required, positional, nargs)
required = self._determine_required(required, positional, nargs, action)
if not isinstance(lazy_resolver, bool):
raise CommandArgumentError(
f"lazy_resolver must be a boolean, got {type(lazy_resolver)}"
)
argument = Argument(
flags=flags,
dest=dest,
action=action,
type=expected_type,
default=default,
choices=choices,
required=required,
help=help,
nargs=nargs,
positional=positional,
resolver=resolver,
lazy_resolver=lazy_resolver,
)
for flag in flags:
if flag in self._flag_map:
existing = self._flag_map[flag]
raise CommandArgumentError(
f"Flag '{flag}' is already used by argument '{existing.dest}'"
)
for flag in flags:
self._flag_map[flag] = argument
if not positional:
self._keyword[flag] = argument
self._dest_set.add(dest)
self._arguments.append(argument)
if positional:
self._positional[dest] = argument
if action == ArgumentAction.STORE_BOOL_OPTIONAL:
self._register_store_bool_optional(flags, dest, help)
else:
self._keyword_list.append(argument)
argument = Argument(
flags=flags,
dest=dest,
action=action,
type=expected_type,
default=default,
choices=choices,
required=required,
help=help,
nargs=nargs,
positional=positional,
resolver=resolver,
lazy_resolver=lazy_resolver,
)
self._register_argument(argument)
def get_argument(self, dest: str) -> Argument | None:
return next((a for a in self._arguments if a.dest == dest), None)
@ -624,6 +707,10 @@ class CommandArgumentParser:
result[spec.dest] = False
consumed_indices.add(i)
i += 1
elif action == ArgumentAction.STORE_BOOL_OPTIONAL:
result[spec.dest] = spec.type(True)
consumed_indices.add(i)
i += 1
elif action == ArgumentAction.COUNT:
result[spec.dest] = result.get(spec.dest, 0) + 1
consumed_indices.add(i)
@ -889,11 +976,28 @@ class CommandArgumentParser:
help_text = f"\n{'':<33}{help_text}"
self.console.print(f"{arg_line}{help_text}")
self.console.print("[bold]options:[/bold]")
arg_groups = defaultdict(list)
for arg in self._keyword_list:
flags = ", ".join(arg.flags)
flags_choice = f"{flags} {arg.get_choice_text()}"
arg_groups[arg.dest].append(arg)
for group in arg_groups.values():
if len(group) == 2 and all(
arg.action == ArgumentAction.STORE_BOOL_OPTIONAL for arg in group
):
# Merge --flag / --no-flag pair into single help line for STORE_BOOL_OPTIONAL
all_flags = tuple(
sorted(
(arg.flags[0] for arg in group),
key=lambda f: f.startswith("--no-"),
)
)
else:
all_flags = group[0].flags
flags = ", ".join(all_flags)
flags_choice = f"{flags} {group[0].get_choice_text()}"
arg_line = f" {flags_choice:<30} "
help_text = arg.help or ""
help_text = group[0].help or ""
if help_text and len(flags_choice) > 30:
help_text = f"\n{'':<33}{help_text}"
self.console.print(f"{arg_line}{help_text}")

15
falyx/parser/parser_types.py Normal file
View File

@ -0,0 +1,15 @@
# Falyx CLI Framework — (c) 2025 rtj.dev LLC — MIT Licensed
"""parser_types.py"""
from typing import Any
def true_none(value: Any) -> bool | None:
if value is None:
return None
return True
def false_none(value: Any) -> bool | None:
if value is None:
return None
return False

4
falyx/parser/utils.py
View File

@ -15,9 +15,9 @@ def coerce_bool(value: str) -> bool:
if isinstance(value, bool):
return value
value = value.strip().lower()
if value in {"true", "1", "yes", "on"}:
if value in {"true", "t", "1", "yes", "on"}:
return True
elif value in {"false", "0", "no", "off"}:
elif value in {"false", "f", "0", "no", "off"}:
return False
return bool(value)

7
falyx/signals.py
View File

@ -10,6 +10,13 @@ class FlowSignal(BaseException):
"""
class BreakChainSignal(FlowSignal):
"""Raised to break the current action chain and return to the previous context."""
def __init__(self, message: str = "Break chain signal received."):
super().__init__(message)
class QuitSignal(FlowSignal):
"""Raised to signal an immediate exit from the CLI framework."""

2
falyx/version.py
View File

@ -1 +1 @@
__version__ = "0.1.59"
__version__ = "0.1.60"

2
pyproject.toml
View File

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

16
tests/test_command.py
View File

@ -98,13 +98,17 @@ def test_enable_retry_not_action():
cmd = Command(
key="C",
description="Retry action",
action=DummyInputAction,
action=DummyInputAction(
name="dummy_input_action",
),
retry=True,
)
assert cmd.retry is True
with pytest.raises(Exception) as exc_info:
assert cmd.action.retry_policy.enabled is False
assert "'function' object has no attribute 'retry_policy'" in str(exc_info.value)
assert "'DummyInputAction' object has no attribute 'retry_policy'" in str(
exc_info.value
)
def test_chain_retry_all():
@ -134,13 +138,17 @@ def test_chain_retry_all_not_base_action():
cmd = Command(
key="E",
description="Chain with retry",
action=DummyInputAction,
action=DummyInputAction(
name="dummy_input_action",
),
retry_all=True,
)
assert cmd.retry_all is True
with pytest.raises(Exception) as exc_info:
assert cmd.action.retry_policy.enabled is False
assert "'function' object has no attribute 'retry_policy'" in str(exc_info.value)
assert "'DummyInputAction' object has no attribute 'retry_policy'" in str(
exc_info.value
)
@pytest.mark.asyncio

10
tests/test_parsers/test_action.py
View File

@ -1,6 +1,6 @@
import pytest
from falyx.action import Action, SelectionAction
from falyx.action import Action
from falyx.exceptions import CommandArgumentError
from falyx.parser import ArgumentAction, CommandArgumentParser
@ -217,11 +217,3 @@ async def test_action_with_default_and_value_positional():
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"])

2
tests/test_parsers/test_argument_action.py
View File

@ -8,4 +8,4 @@ def test_argument_action():
assert action != "invalid_action"
assert action.value == "append"
assert str(action) == "append"
assert len(ArgumentAction.choices()) == 8
assert len(ArgumentAction.choices()) == 9

83
tests/test_parsers/test_store_bool_optional.py Normal file
View File

@ -0,0 +1,83 @@
import pytest
from falyx.exceptions import CommandArgumentError
from falyx.parser import ArgumentAction, CommandArgumentParser
@pytest.mark.asyncio
async def test_store_bool_optional_true():
parser = CommandArgumentParser()
parser.add_argument(
"--debug",
action=ArgumentAction.STORE_BOOL_OPTIONAL,
help="Enable debug mode.",
)
args = await parser.parse_args(["--debug"])
assert args["debug"] is True
@pytest.mark.asyncio
async def test_store_bool_optional_false():
parser = CommandArgumentParser()
parser.add_argument(
"--debug",
action=ArgumentAction.STORE_BOOL_OPTIONAL,
help="Enable debug mode.",
)
args = await parser.parse_args(["--no-debug"])
assert args["debug"] is False
@pytest.mark.asyncio
async def test_store_bool_optional_default_none():
parser = CommandArgumentParser()
parser.add_argument(
"--debug",
action=ArgumentAction.STORE_BOOL_OPTIONAL,
help="Enable debug mode.",
)
args = await parser.parse_args([])
assert args["debug"] is None
@pytest.mark.asyncio
async def test_store_bool_optional_flag_order():
parser = CommandArgumentParser()
parser.add_argument(
"--dry-run",
action=ArgumentAction.STORE_BOOL_OPTIONAL,
help="Run without making changes.",
)
args = await parser.parse_args(["--dry-run"])
assert args["dry_run"] is True
args = await parser.parse_args(["--no-dry-run"])
assert args["dry_run"] is False
def test_store_bool_optional_requires_long_flag():
parser = CommandArgumentParser()
with pytest.raises(CommandArgumentError):
parser.add_argument(
"-d", action=ArgumentAction.STORE_BOOL_OPTIONAL, help="Invalid"
)
def test_store_bool_optional_disallows_multiple_flags():
parser = CommandArgumentParser()
with pytest.raises(CommandArgumentError):
parser.add_argument("--debug", "-d", action=ArgumentAction.STORE_BOOL_OPTIONAL)
def test_store_bool_optional_duplicate_dest():
parser = CommandArgumentParser()
parser.add_argument(
"--debug",
action=ArgumentAction.STORE_BOOL_OPTIONAL,
help="Enable debug mode.",
)
with pytest.raises(CommandArgumentError):
parser.add_argument(
"--debug",
action=ArgumentAction.STORE_TRUE,
help="Conflicting debug option.",
)