Merge pull request 'Add auto_args' (#3) from argparse-integration into main
Reviewed-on: #3
This commit is contained in:
commit
4fa6e3bf1f
|
@ -0,0 +1,40 @@
|
||||||
|
import asyncio
|
||||||
|
|
||||||
|
from falyx import Action, ActionGroup, Command, Falyx
|
||||||
|
|
||||||
|
|
||||||
|
# Define a shared async function
|
||||||
|
async def say_hello(name: str, excited: bool = False):
|
||||||
|
if excited:
|
||||||
|
print(f"Hello, {name}!!!")
|
||||||
|
else:
|
||||||
|
print(f"Hello, {name}.")
|
||||||
|
|
||||||
|
|
||||||
|
# Wrap the same callable in multiple Actions
|
||||||
|
action1 = Action("say_hello_1", action=say_hello)
|
||||||
|
action2 = Action("say_hello_2", action=say_hello)
|
||||||
|
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(
|
||||||
|
key="G",
|
||||||
|
description="Greet someone with multiple variations.",
|
||||||
|
action=group,
|
||||||
|
auto_args=True,
|
||||||
|
arg_metadata={
|
||||||
|
"name": {
|
||||||
|
"help": "The name of the person to greet.",
|
||||||
|
},
|
||||||
|
"excited": {
|
||||||
|
"help": "Whether to greet excitedly.",
|
||||||
|
},
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
flx = Falyx("Test Group")
|
||||||
|
flx.add_command_from_command(cmd)
|
||||||
|
asyncio.run(flx.run())
|
|
@ -0,0 +1,32 @@
|
||||||
|
import asyncio
|
||||||
|
|
||||||
|
from falyx import Action, Falyx
|
||||||
|
|
||||||
|
|
||||||
|
async def deploy(service: str, region: str = "us-east-1", verbose: bool = False):
|
||||||
|
if verbose:
|
||||||
|
print(f"Deploying {service} to {region}...")
|
||||||
|
await asyncio.sleep(2)
|
||||||
|
if verbose:
|
||||||
|
print(f"{service} deployed successfully!")
|
||||||
|
|
||||||
|
|
||||||
|
flx = Falyx("Deployment CLI")
|
||||||
|
|
||||||
|
flx.add_command(
|
||||||
|
key="D",
|
||||||
|
aliases=["deploy"],
|
||||||
|
description="Deploy a service to a specified region.",
|
||||||
|
action=Action(
|
||||||
|
name="deploy_service",
|
||||||
|
action=deploy,
|
||||||
|
),
|
||||||
|
auto_args=True,
|
||||||
|
arg_metadata={
|
||||||
|
"service": "Service name",
|
||||||
|
"region": {"help": "Deployment region", "choices": ["us-east-1", "us-west-2"]},
|
||||||
|
"verbose": {"help": "Enable verbose mode"},
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
asyncio.run(flx.run())
|
|
@ -224,7 +224,10 @@ class ShellAction(BaseIOAction):
|
||||||
# Replace placeholder in template, or use raw input as full command
|
# Replace placeholder in template, or use raw input as full command
|
||||||
command = self.command_template.format(parsed_input)
|
command = self.command_template.format(parsed_input)
|
||||||
if self.safe_mode:
|
if self.safe_mode:
|
||||||
|
try:
|
||||||
args = shlex.split(command)
|
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)
|
result = subprocess.run(args, capture_output=True, text=True, check=True)
|
||||||
else:
|
else:
|
||||||
result = subprocess.run(
|
result = subprocess.run(
|
||||||
|
|
|
@ -27,15 +27,25 @@ from pydantic import BaseModel, ConfigDict, Field, PrivateAttr, field_validator
|
||||||
from rich.console import Console
|
from rich.console import Console
|
||||||
from rich.tree import Tree
|
from rich.tree import Tree
|
||||||
|
|
||||||
from falyx.action.action import Action, ActionGroup, BaseAction, ChainedAction
|
from falyx.action.action import (
|
||||||
|
Action,
|
||||||
|
ActionGroup,
|
||||||
|
BaseAction,
|
||||||
|
ChainedAction,
|
||||||
|
ProcessAction,
|
||||||
|
)
|
||||||
from falyx.action.io_action import BaseIOAction
|
from falyx.action.io_action import BaseIOAction
|
||||||
from falyx.argparse import CommandArgumentParser
|
|
||||||
from falyx.context import ExecutionContext
|
from falyx.context import ExecutionContext
|
||||||
from falyx.debug import register_debug_hooks
|
from falyx.debug import register_debug_hooks
|
||||||
from falyx.execution_registry import ExecutionRegistry as er
|
from falyx.execution_registry import ExecutionRegistry as er
|
||||||
from falyx.hook_manager import HookManager, HookType
|
from falyx.hook_manager import HookManager, HookType
|
||||||
from falyx.logger import logger
|
from falyx.logger import logger
|
||||||
from falyx.options_manager import OptionsManager
|
from falyx.options_manager import OptionsManager
|
||||||
|
from falyx.parsers import (
|
||||||
|
CommandArgumentParser,
|
||||||
|
infer_args_from_func,
|
||||||
|
same_argument_definitions,
|
||||||
|
)
|
||||||
from falyx.prompt_utils import confirm_async, should_prompt_user
|
from falyx.prompt_utils import confirm_async, should_prompt_user
|
||||||
from falyx.protocols import ArgParserProtocol
|
from falyx.protocols import ArgParserProtocol
|
||||||
from falyx.retry import RetryPolicy
|
from falyx.retry import RetryPolicy
|
||||||
|
@ -90,6 +100,11 @@ class Command(BaseModel):
|
||||||
tags (list[str]): Organizational tags for the command.
|
tags (list[str]): Organizational tags for the command.
|
||||||
logging_hooks (bool): Whether to attach logging hooks automatically.
|
logging_hooks (bool): Whether to attach logging hooks automatically.
|
||||||
requires_input (bool | None): Indicates if the action needs input.
|
requires_input (bool | None): Indicates if the action needs input.
|
||||||
|
options_manager (OptionsManager): Manages global command-line options.
|
||||||
|
arg_parser (CommandArgumentParser): Parses command arguments.
|
||||||
|
custom_parser (ArgParserProtocol | None): Custom argument parser.
|
||||||
|
custom_help (Callable[[], str | None] | None): Custom help message generator.
|
||||||
|
auto_args (bool): Automatically infer arguments from the action.
|
||||||
|
|
||||||
Methods:
|
Methods:
|
||||||
__call__(): Executes the command, respecting hooks and retries.
|
__call__(): Executes the command, respecting hooks and retries.
|
||||||
|
@ -101,12 +116,13 @@ class Command(BaseModel):
|
||||||
|
|
||||||
key: str
|
key: str
|
||||||
description: str
|
description: str
|
||||||
action: BaseAction | Callable[[], Any]
|
action: BaseAction | Callable[[Any], Any]
|
||||||
args: tuple = ()
|
args: tuple = ()
|
||||||
kwargs: dict[str, Any] = Field(default_factory=dict)
|
kwargs: dict[str, Any] = Field(default_factory=dict)
|
||||||
hidden: bool = False
|
hidden: bool = False
|
||||||
aliases: list[str] = Field(default_factory=list)
|
aliases: list[str] = Field(default_factory=list)
|
||||||
help_text: str = ""
|
help_text: str = ""
|
||||||
|
help_epilogue: str = ""
|
||||||
style: str = OneColors.WHITE
|
style: str = OneColors.WHITE
|
||||||
confirm: bool = False
|
confirm: bool = False
|
||||||
confirm_message: str = "Are you sure?"
|
confirm_message: str = "Are you sure?"
|
||||||
|
@ -125,22 +141,44 @@ class Command(BaseModel):
|
||||||
requires_input: bool | None = None
|
requires_input: bool | None = None
|
||||||
options_manager: OptionsManager = Field(default_factory=OptionsManager)
|
options_manager: OptionsManager = Field(default_factory=OptionsManager)
|
||||||
arg_parser: CommandArgumentParser = Field(default_factory=CommandArgumentParser)
|
arg_parser: CommandArgumentParser = Field(default_factory=CommandArgumentParser)
|
||||||
|
arguments: list[dict[str, Any]] = Field(default_factory=list)
|
||||||
|
argument_config: Callable[[CommandArgumentParser], None] | None = None
|
||||||
custom_parser: ArgParserProtocol | None = None
|
custom_parser: ArgParserProtocol | None = None
|
||||||
custom_help: Callable[[], str | None] | None = None
|
custom_help: Callable[[], str | None] | None = None
|
||||||
|
auto_args: bool = False
|
||||||
|
arg_metadata: dict[str, str | dict[str, Any]] = Field(default_factory=dict)
|
||||||
|
|
||||||
_context: ExecutionContext | None = PrivateAttr(default=None)
|
_context: ExecutionContext | None = PrivateAttr(default=None)
|
||||||
|
|
||||||
model_config = ConfigDict(arbitrary_types_allowed=True)
|
model_config = ConfigDict(arbitrary_types_allowed=True)
|
||||||
|
|
||||||
def parse_args(self, raw_args: list[str] | str) -> tuple[tuple, dict]:
|
def parse_args(
|
||||||
|
self, raw_args: list[str] | str, from_validate: bool = False
|
||||||
|
) -> tuple[tuple, dict]:
|
||||||
if self.custom_parser:
|
if self.custom_parser:
|
||||||
if isinstance(raw_args, str):
|
if isinstance(raw_args, str):
|
||||||
|
try:
|
||||||
raw_args = shlex.split(raw_args)
|
raw_args = shlex.split(raw_args)
|
||||||
|
except ValueError:
|
||||||
|
logger.warning(
|
||||||
|
"[Command:%s] Failed to split arguments: %s",
|
||||||
|
self.key,
|
||||||
|
raw_args,
|
||||||
|
)
|
||||||
|
return ((), {})
|
||||||
return self.custom_parser(raw_args)
|
return self.custom_parser(raw_args)
|
||||||
|
|
||||||
if isinstance(raw_args, str):
|
if isinstance(raw_args, str):
|
||||||
|
try:
|
||||||
raw_args = shlex.split(raw_args)
|
raw_args = shlex.split(raw_args)
|
||||||
return self.arg_parser.parse_args_split(raw_args)
|
except ValueError:
|
||||||
|
logger.warning(
|
||||||
|
"[Command:%s] Failed to split arguments: %s",
|
||||||
|
self.key,
|
||||||
|
raw_args,
|
||||||
|
)
|
||||||
|
return ((), {})
|
||||||
|
return self.arg_parser.parse_args_split(raw_args, from_validate=from_validate)
|
||||||
|
|
||||||
@field_validator("action", mode="before")
|
@field_validator("action", mode="before")
|
||||||
@classmethod
|
@classmethod
|
||||||
|
@ -151,11 +189,37 @@ class Command(BaseModel):
|
||||||
return ensure_async(action)
|
return ensure_async(action)
|
||||||
raise TypeError("Action must be a callable or an instance of BaseAction")
|
raise TypeError("Action must be a callable or an instance of BaseAction")
|
||||||
|
|
||||||
|
def get_argument_definitions(self) -> list[dict[str, Any]]:
|
||||||
|
if self.arguments:
|
||||||
|
return self.arguments
|
||||||
|
elif self.argument_config:
|
||||||
|
self.argument_config(self.arg_parser)
|
||||||
|
elif self.auto_args:
|
||||||
|
if isinstance(self.action, (Action, ProcessAction)):
|
||||||
|
return infer_args_from_func(self.action.action, self.arg_metadata)
|
||||||
|
elif isinstance(self.action, ChainedAction):
|
||||||
|
if self.action.actions:
|
||||||
|
action = self.action.actions[0]
|
||||||
|
if isinstance(action, Action):
|
||||||
|
return infer_args_from_func(action.action, self.arg_metadata)
|
||||||
|
elif callable(action):
|
||||||
|
return infer_args_from_func(action, self.arg_metadata)
|
||||||
|
elif isinstance(self.action, ActionGroup):
|
||||||
|
arg_defs = same_argument_definitions(
|
||||||
|
self.action.actions, self.arg_metadata
|
||||||
|
)
|
||||||
|
if arg_defs:
|
||||||
|
return arg_defs
|
||||||
|
logger.debug(
|
||||||
|
"[Command:%s] auto_args disabled: mismatched ActionGroup arguments",
|
||||||
|
self.key,
|
||||||
|
)
|
||||||
|
elif callable(self.action):
|
||||||
|
return infer_args_from_func(self.action, self.arg_metadata)
|
||||||
|
return []
|
||||||
|
|
||||||
def model_post_init(self, _: Any) -> None:
|
def model_post_init(self, _: Any) -> None:
|
||||||
"""Post-initialization to set up the action and hooks."""
|
"""Post-initialization to set up the action and hooks."""
|
||||||
if isinstance(self.arg_parser, CommandArgumentParser):
|
|
||||||
self.arg_parser.command_description = self.description
|
|
||||||
|
|
||||||
if self.retry and isinstance(self.action, Action):
|
if self.retry and isinstance(self.action, Action):
|
||||||
self.action.enable_retry()
|
self.action.enable_retry()
|
||||||
elif self.retry_policy and isinstance(self.action, Action):
|
elif self.retry_policy and isinstance(self.action, Action):
|
||||||
|
@ -183,6 +247,9 @@ class Command(BaseModel):
|
||||||
elif self.requires_input is None:
|
elif self.requires_input is None:
|
||||||
self.requires_input = False
|
self.requires_input = False
|
||||||
|
|
||||||
|
for arg_def in self.get_argument_definitions():
|
||||||
|
self.arg_parser.add_argument(*arg_def.pop("flags"), **arg_def)
|
||||||
|
|
||||||
@cached_property
|
@cached_property
|
||||||
def detect_requires_input(self) -> bool:
|
def detect_requires_input(self) -> bool:
|
||||||
"""Detect if the action requires input based on its type."""
|
"""Detect if the action requires input based on its type."""
|
||||||
|
|
|
@ -58,9 +58,10 @@ from falyx.execution_registry import ExecutionRegistry as er
|
||||||
from falyx.hook_manager import Hook, HookManager, HookType
|
from falyx.hook_manager import Hook, HookManager, HookType
|
||||||
from falyx.logger import logger
|
from falyx.logger import logger
|
||||||
from falyx.options_manager import OptionsManager
|
from falyx.options_manager import OptionsManager
|
||||||
from falyx.parsers import get_arg_parsers
|
from falyx.parsers import CommandArgumentParser, get_arg_parsers
|
||||||
|
from falyx.protocols import ArgParserProtocol
|
||||||
from falyx.retry import RetryPolicy
|
from falyx.retry import RetryPolicy
|
||||||
from falyx.signals import BackSignal, CancelSignal, HelpSignal, QuitSignal
|
from falyx.signals import BackSignal, CancelSignal, FlowSignal, HelpSignal, QuitSignal
|
||||||
from falyx.themes import OneColors, get_nord_theme
|
from falyx.themes import OneColors, get_nord_theme
|
||||||
from falyx.utils import CaseInsensitiveDict, _noop, chunks, get_program_invocation
|
from falyx.utils import CaseInsensitiveDict, _noop, chunks, get_program_invocation
|
||||||
from falyx.version import __version__
|
from falyx.version import __version__
|
||||||
|
@ -444,6 +445,7 @@ class Falyx:
|
||||||
bottom_toolbar=self._get_bottom_bar_render(),
|
bottom_toolbar=self._get_bottom_bar_render(),
|
||||||
key_bindings=self.key_bindings,
|
key_bindings=self.key_bindings,
|
||||||
validate_while_typing=False,
|
validate_while_typing=False,
|
||||||
|
interrupt_exception=FlowSignal,
|
||||||
)
|
)
|
||||||
return self._prompt_session
|
return self._prompt_session
|
||||||
|
|
||||||
|
@ -524,7 +526,7 @@ class Falyx:
|
||||||
key: str = "X",
|
key: str = "X",
|
||||||
description: str = "Exit",
|
description: str = "Exit",
|
||||||
aliases: list[str] | None = None,
|
aliases: list[str] | None = None,
|
||||||
action: Callable[[], Any] | None = None,
|
action: Callable[[Any], Any] | None = None,
|
||||||
style: str = OneColors.DARK_RED,
|
style: str = OneColors.DARK_RED,
|
||||||
confirm: bool = False,
|
confirm: bool = False,
|
||||||
confirm_message: str = "Are you sure?",
|
confirm_message: str = "Are you sure?",
|
||||||
|
@ -578,13 +580,14 @@ class Falyx:
|
||||||
self,
|
self,
|
||||||
key: str,
|
key: str,
|
||||||
description: str,
|
description: str,
|
||||||
action: BaseAction | Callable[[], Any],
|
action: BaseAction | Callable[[Any], Any],
|
||||||
*,
|
*,
|
||||||
args: tuple = (),
|
args: tuple = (),
|
||||||
kwargs: dict[str, Any] | None = None,
|
kwargs: dict[str, Any] | None = None,
|
||||||
hidden: bool = False,
|
hidden: bool = False,
|
||||||
aliases: list[str] | None = None,
|
aliases: list[str] | None = None,
|
||||||
help_text: str = "",
|
help_text: str = "",
|
||||||
|
help_epilogue: str = "",
|
||||||
style: str = OneColors.WHITE,
|
style: str = OneColors.WHITE,
|
||||||
confirm: bool = False,
|
confirm: bool = False,
|
||||||
confirm_message: str = "Are you sure?",
|
confirm_message: str = "Are you sure?",
|
||||||
|
@ -606,9 +609,33 @@ class Falyx:
|
||||||
retry_all: bool = False,
|
retry_all: bool = False,
|
||||||
retry_policy: RetryPolicy | None = None,
|
retry_policy: RetryPolicy | None = None,
|
||||||
requires_input: bool | None = None,
|
requires_input: bool | None = None,
|
||||||
|
arg_parser: CommandArgumentParser | None = None,
|
||||||
|
arguments: list[dict[str, Any]] | None = None,
|
||||||
|
argument_config: Callable[[CommandArgumentParser], None] | None = None,
|
||||||
|
custom_parser: ArgParserProtocol | None = None,
|
||||||
|
custom_help: Callable[[], str | None] | None = None,
|
||||||
|
auto_args: bool = False,
|
||||||
|
arg_metadata: dict[str, str | dict[str, Any]] | None = None,
|
||||||
) -> Command:
|
) -> Command:
|
||||||
"""Adds an command to the menu, preventing duplicates."""
|
"""Adds an command to the menu, preventing duplicates."""
|
||||||
self._validate_command_key(key)
|
self._validate_command_key(key)
|
||||||
|
|
||||||
|
if arg_parser:
|
||||||
|
if not isinstance(arg_parser, CommandArgumentParser):
|
||||||
|
raise NotAFalyxError(
|
||||||
|
"arg_parser must be an instance of CommandArgumentParser."
|
||||||
|
)
|
||||||
|
arg_parser = arg_parser
|
||||||
|
else:
|
||||||
|
arg_parser = CommandArgumentParser(
|
||||||
|
command_key=key,
|
||||||
|
command_description=description,
|
||||||
|
command_style=style,
|
||||||
|
help_text=help_text,
|
||||||
|
help_epilogue=help_epilogue,
|
||||||
|
aliases=aliases,
|
||||||
|
)
|
||||||
|
|
||||||
command = Command(
|
command = Command(
|
||||||
key=key,
|
key=key,
|
||||||
description=description,
|
description=description,
|
||||||
|
@ -618,6 +645,7 @@ class Falyx:
|
||||||
hidden=hidden,
|
hidden=hidden,
|
||||||
aliases=aliases if aliases else [],
|
aliases=aliases if aliases else [],
|
||||||
help_text=help_text,
|
help_text=help_text,
|
||||||
|
help_epilogue=help_epilogue,
|
||||||
style=style,
|
style=style,
|
||||||
confirm=confirm,
|
confirm=confirm,
|
||||||
confirm_message=confirm_message,
|
confirm_message=confirm_message,
|
||||||
|
@ -634,6 +662,13 @@ class Falyx:
|
||||||
retry_policy=retry_policy or RetryPolicy(),
|
retry_policy=retry_policy or RetryPolicy(),
|
||||||
requires_input=requires_input,
|
requires_input=requires_input,
|
||||||
options_manager=self.options,
|
options_manager=self.options,
|
||||||
|
arg_parser=arg_parser,
|
||||||
|
arguments=arguments or [],
|
||||||
|
argument_config=argument_config,
|
||||||
|
custom_parser=custom_parser,
|
||||||
|
custom_help=custom_help,
|
||||||
|
auto_args=auto_args,
|
||||||
|
arg_metadata=arg_metadata or {},
|
||||||
)
|
)
|
||||||
|
|
||||||
if hooks:
|
if hooks:
|
||||||
|
@ -715,7 +750,10 @@ class Falyx:
|
||||||
"""
|
"""
|
||||||
args = ()
|
args = ()
|
||||||
kwargs: dict[str, Any] = {}
|
kwargs: dict[str, Any] = {}
|
||||||
|
try:
|
||||||
choice, *input_args = shlex.split(raw_choices)
|
choice, *input_args = shlex.split(raw_choices)
|
||||||
|
except ValueError:
|
||||||
|
return False, None, args, kwargs
|
||||||
is_preview, choice = self.parse_preview_command(choice)
|
is_preview, choice = self.parse_preview_command(choice)
|
||||||
if is_preview and not choice and self.help_command:
|
if is_preview and not choice and self.help_command:
|
||||||
is_preview = False
|
is_preview = False
|
||||||
|
@ -735,7 +773,7 @@ class Falyx:
|
||||||
logger.info("Command '%s' selected.", choice)
|
logger.info("Command '%s' selected.", choice)
|
||||||
if input_args and name_map[choice].arg_parser:
|
if input_args and name_map[choice].arg_parser:
|
||||||
try:
|
try:
|
||||||
args, kwargs = name_map[choice].parse_args(input_args)
|
args, kwargs = name_map[choice].parse_args(input_args, from_validate)
|
||||||
except CommandArgumentError as error:
|
except CommandArgumentError as error:
|
||||||
if not from_validate:
|
if not from_validate:
|
||||||
if not name_map[choice].show_help():
|
if not name_map[choice].show_help():
|
||||||
|
@ -748,6 +786,8 @@ class Falyx:
|
||||||
message=str(error), cursor_position=len(raw_choices)
|
message=str(error), cursor_position=len(raw_choices)
|
||||||
)
|
)
|
||||||
return is_preview, None, args, kwargs
|
return is_preview, None, args, kwargs
|
||||||
|
except HelpSignal:
|
||||||
|
return True, None, args, kwargs
|
||||||
return is_preview, name_map[choice], args, kwargs
|
return is_preview, name_map[choice], args, kwargs
|
||||||
|
|
||||||
prefix_matches = [cmd for key, cmd in name_map.items() if key.startswith(choice)]
|
prefix_matches = [cmd for key, cmd in name_map.items() if key.startswith(choice)]
|
||||||
|
@ -823,7 +863,6 @@ class Falyx:
|
||||||
context.start_timer()
|
context.start_timer()
|
||||||
try:
|
try:
|
||||||
await self.hooks.trigger(HookType.BEFORE, context)
|
await self.hooks.trigger(HookType.BEFORE, context)
|
||||||
print(args, kwargs)
|
|
||||||
result = await selected_command(*args, **kwargs)
|
result = await selected_command(*args, **kwargs)
|
||||||
context.result = result
|
context.result = result
|
||||||
await self.hooks.trigger(HookType.ON_SUCCESS, context)
|
await self.hooks.trigger(HookType.ON_SUCCESS, context)
|
||||||
|
@ -964,8 +1003,6 @@ class Falyx:
|
||||||
logger.info("BackSignal received.")
|
logger.info("BackSignal received.")
|
||||||
except CancelSignal:
|
except CancelSignal:
|
||||||
logger.info("CancelSignal received.")
|
logger.info("CancelSignal received.")
|
||||||
except HelpSignal:
|
|
||||||
logger.info("HelpSignal received.")
|
|
||||||
finally:
|
finally:
|
||||||
logger.info("Exiting menu: %s", self.get_title())
|
logger.info("Exiting menu: %s", self.get_title())
|
||||||
if self.exit_message:
|
if self.exit_message:
|
||||||
|
@ -995,7 +1032,7 @@ class Falyx:
|
||||||
sys.exit(0)
|
sys.exit(0)
|
||||||
|
|
||||||
if self.cli_args.command == "version" or self.cli_args.version:
|
if self.cli_args.command == "version" or self.cli_args.version:
|
||||||
self.console.print(f"[{OneColors.GREEN_b}]Falyx CLI v{__version__}[/]")
|
self.console.print(f"[{OneColors.BLUE_b}]Falyx CLI v{__version__}[/]")
|
||||||
sys.exit(0)
|
sys.exit(0)
|
||||||
|
|
||||||
if self.cli_args.command == "preview":
|
if self.cli_args.command == "preview":
|
||||||
|
|
|
@ -0,0 +1,21 @@
|
||||||
|
"""
|
||||||
|
Falyx CLI Framework
|
||||||
|
|
||||||
|
Copyright (c) 2025 rtj.dev LLC.
|
||||||
|
Licensed under the MIT License. See LICENSE file for details.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from .argparse import Argument, ArgumentAction, CommandArgumentParser
|
||||||
|
from .parsers import FalyxParsers, get_arg_parsers
|
||||||
|
from .signature import infer_args_from_func
|
||||||
|
from .utils import same_argument_definitions
|
||||||
|
|
||||||
|
__all__ = [
|
||||||
|
"Argument",
|
||||||
|
"ArgumentAction",
|
||||||
|
"CommandArgumentParser",
|
||||||
|
"get_arg_parsers",
|
||||||
|
"FalyxParsers",
|
||||||
|
"infer_args_from_func",
|
||||||
|
"same_argument_definitions",
|
||||||
|
]
|
|
@ -5,7 +5,8 @@ from enum import Enum
|
||||||
from typing import Any, Iterable
|
from typing import Any, Iterable
|
||||||
|
|
||||||
from rich.console import Console
|
from rich.console import Console
|
||||||
from rich.table import Table
|
from rich.markup import escape
|
||||||
|
from rich.text import Text
|
||||||
|
|
||||||
from falyx.exceptions import CommandArgumentError
|
from falyx.exceptions import CommandArgumentError
|
||||||
from falyx.signals import HelpSignal
|
from falyx.signals import HelpSignal
|
||||||
|
@ -40,6 +41,70 @@ class Argument:
|
||||||
nargs: int | str = 1 # int, '?', '*', '+'
|
nargs: int | str = 1 # int, '?', '*', '+'
|
||||||
positional: bool = False # True if no leading - or -- in flags
|
positional: bool = False # True if no leading - or -- in flags
|
||||||
|
|
||||||
|
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 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
|
||||||
|
)
|
||||||
|
|
||||||
|
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,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
class CommandArgumentParser:
|
class CommandArgumentParser:
|
||||||
"""
|
"""
|
||||||
|
@ -61,10 +126,25 @@ class CommandArgumentParser:
|
||||||
- Render Help using Rich library.
|
- Render Help using Rich library.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self) -> None:
|
def __init__(
|
||||||
|
self,
|
||||||
|
command_key: str = "",
|
||||||
|
command_description: str = "",
|
||||||
|
command_style: str = "bold",
|
||||||
|
help_text: str = "",
|
||||||
|
help_epilogue: str = "",
|
||||||
|
aliases: list[str] | None = None,
|
||||||
|
) -> None:
|
||||||
"""Initialize the CommandArgumentParser."""
|
"""Initialize the CommandArgumentParser."""
|
||||||
self.command_description: str = ""
|
self.command_key: str = command_key
|
||||||
|
self.command_description: str = command_description
|
||||||
|
self.command_style: str = command_style
|
||||||
|
self.help_text: str = help_text
|
||||||
|
self.help_epilogue: str = help_epilogue
|
||||||
|
self.aliases: list[str] = aliases or []
|
||||||
self._arguments: list[Argument] = []
|
self._arguments: list[Argument] = []
|
||||||
|
self._positional: list[Argument] = []
|
||||||
|
self._keyword: list[Argument] = []
|
||||||
self._flag_map: dict[str, Argument] = {}
|
self._flag_map: dict[str, Argument] = {}
|
||||||
self._dest_set: set[str] = set()
|
self._dest_set: set[str] = set()
|
||||||
self._add_help()
|
self._add_help()
|
||||||
|
@ -73,10 +153,10 @@ class CommandArgumentParser:
|
||||||
def _add_help(self):
|
def _add_help(self):
|
||||||
"""Add help argument to the parser."""
|
"""Add help argument to the parser."""
|
||||||
self.add_argument(
|
self.add_argument(
|
||||||
"--help",
|
|
||||||
"-h",
|
"-h",
|
||||||
|
"--help",
|
||||||
action=ArgumentAction.HELP,
|
action=ArgumentAction.HELP,
|
||||||
help="Show this help message and exit.",
|
help="Show this help message.",
|
||||||
dest="help",
|
dest="help",
|
||||||
)
|
)
|
||||||
|
|
||||||
|
@ -304,10 +384,31 @@ class CommandArgumentParser:
|
||||||
)
|
)
|
||||||
self._flag_map[flag] = argument
|
self._flag_map[flag] = argument
|
||||||
self._arguments.append(argument)
|
self._arguments.append(argument)
|
||||||
|
if positional:
|
||||||
|
self._positional.append(argument)
|
||||||
|
else:
|
||||||
|
self._keyword.append(argument)
|
||||||
|
|
||||||
def get_argument(self, dest: str) -> Argument | None:
|
def get_argument(self, dest: str) -> Argument | None:
|
||||||
return next((a for a in self._arguments if a.dest == dest), None)
|
return next((a for a in self._arguments if a.dest == dest), None)
|
||||||
|
|
||||||
|
def to_definition_list(self) -> list[dict[str, Any]]:
|
||||||
|
defs = []
|
||||||
|
for arg in self._arguments:
|
||||||
|
defs.append(
|
||||||
|
{
|
||||||
|
"flags": arg.flags,
|
||||||
|
"dest": arg.dest,
|
||||||
|
"action": arg.action,
|
||||||
|
"type": arg.type,
|
||||||
|
"choices": arg.choices,
|
||||||
|
"required": arg.required,
|
||||||
|
"nargs": arg.nargs,
|
||||||
|
"positional": arg.positional,
|
||||||
|
}
|
||||||
|
)
|
||||||
|
return defs
|
||||||
|
|
||||||
def _consume_nargs(
|
def _consume_nargs(
|
||||||
self, args: list[str], start: int, spec: Argument
|
self, args: list[str], start: int, spec: Argument
|
||||||
) -> tuple[list[str], int]:
|
) -> tuple[list[str], int]:
|
||||||
|
@ -405,7 +506,9 @@ class CommandArgumentParser:
|
||||||
|
|
||||||
return i
|
return i
|
||||||
|
|
||||||
def parse_args(self, args: list[str] | None = None) -> dict[str, Any]:
|
def parse_args(
|
||||||
|
self, args: list[str] | None = None, from_validate: bool = False
|
||||||
|
) -> dict[str, Any]:
|
||||||
"""Parse Falyx Command arguments."""
|
"""Parse Falyx Command arguments."""
|
||||||
if args is None:
|
if args is None:
|
||||||
args = []
|
args = []
|
||||||
|
@ -423,6 +526,7 @@ class CommandArgumentParser:
|
||||||
action = spec.action
|
action = spec.action
|
||||||
|
|
||||||
if action == ArgumentAction.HELP:
|
if action == ArgumentAction.HELP:
|
||||||
|
if not from_validate:
|
||||||
self.render_help()
|
self.render_help()
|
||||||
raise HelpSignal()
|
raise HelpSignal()
|
||||||
elif action == ArgumentAction.STORE_TRUE:
|
elif action == ArgumentAction.STORE_TRUE:
|
||||||
|
@ -550,13 +654,15 @@ class CommandArgumentParser:
|
||||||
result.pop("help", None)
|
result.pop("help", None)
|
||||||
return result
|
return result
|
||||||
|
|
||||||
def parse_args_split(self, args: list[str]) -> tuple[tuple[Any, ...], dict[str, Any]]:
|
def parse_args_split(
|
||||||
|
self, args: list[str], from_validate: bool = False
|
||||||
|
) -> tuple[tuple[Any, ...], dict[str, Any]]:
|
||||||
"""
|
"""
|
||||||
Returns:
|
Returns:
|
||||||
tuple[args, kwargs] - Positional arguments in defined order,
|
tuple[args, kwargs] - Positional arguments in defined order,
|
||||||
followed by keyword argument mapping.
|
followed by keyword argument mapping.
|
||||||
"""
|
"""
|
||||||
parsed = self.parse_args(args)
|
parsed = self.parse_args(args, from_validate)
|
||||||
args_list = []
|
args_list = []
|
||||||
kwargs_dict = {}
|
kwargs_dict = {}
|
||||||
for arg in self._arguments:
|
for arg in self._arguments:
|
||||||
|
@ -568,20 +674,74 @@ class CommandArgumentParser:
|
||||||
kwargs_dict[arg.dest] = parsed[arg.dest]
|
kwargs_dict[arg.dest] = parsed[arg.dest]
|
||||||
return tuple(args_list), kwargs_dict
|
return tuple(args_list), kwargs_dict
|
||||||
|
|
||||||
def render_help(self):
|
def render_help(self) -> None:
|
||||||
table = Table(title=f"{self.command_description} Help")
|
# Options
|
||||||
table.add_column("Flags")
|
# Add all keyword arguments to the options list
|
||||||
table.add_column("Help")
|
options_list = []
|
||||||
for arg in self._arguments:
|
for arg in self._keyword:
|
||||||
if arg.dest == "help":
|
choice_text = arg.get_choice_text()
|
||||||
continue
|
if choice_text:
|
||||||
flag_str = ", ".join(arg.flags) if not arg.positional else arg.dest
|
options_list.extend([f"[{arg.flags[0]} {choice_text}]"])
|
||||||
table.add_row(flag_str, arg.help or "")
|
else:
|
||||||
table.add_section()
|
options_list.extend([f"[{arg.flags[0]}]"])
|
||||||
arg = self.get_argument("help")
|
|
||||||
flag_str = ", ".join(arg.flags) if not arg.positional else arg.dest
|
# Add positional arguments to the options list
|
||||||
table.add_row(flag_str, arg.help or "")
|
for arg in self._positional:
|
||||||
self.console.print(table)
|
choice_text = arg.get_choice_text()
|
||||||
|
if isinstance(arg.nargs, int):
|
||||||
|
choice_text = " ".join([choice_text] * arg.nargs)
|
||||||
|
options_list.append(escape(choice_text))
|
||||||
|
|
||||||
|
options_text = " ".join(options_list)
|
||||||
|
command_keys = " | ".join(
|
||||||
|
[f"[{self.command_style}]{self.command_key}[/{self.command_style}]"]
|
||||||
|
+ [
|
||||||
|
f"[{self.command_style}]{alias}[/{self.command_style}]"
|
||||||
|
for alias in self.aliases
|
||||||
|
]
|
||||||
|
)
|
||||||
|
|
||||||
|
usage = f"usage: {command_keys} {options_text}"
|
||||||
|
self.console.print(f"[bold]{usage}[/bold]\n")
|
||||||
|
|
||||||
|
# Description
|
||||||
|
if self.help_text:
|
||||||
|
self.console.print(self.help_text + "\n")
|
||||||
|
|
||||||
|
# Arguments
|
||||||
|
if self._arguments:
|
||||||
|
if self._positional:
|
||||||
|
self.console.print("[bold]positional:[/bold]")
|
||||||
|
for arg in self._positional:
|
||||||
|
flags = arg.get_positional_text()
|
||||||
|
arg_line = Text(f" {flags:<30} ")
|
||||||
|
help_text = arg.help or ""
|
||||||
|
arg_line.append(help_text)
|
||||||
|
self.console.print(arg_line)
|
||||||
|
self.console.print("[bold]options:[/bold]")
|
||||||
|
for arg in self._keyword:
|
||||||
|
flags = ", ".join(arg.flags)
|
||||||
|
flags_choice = f"{flags} {arg.get_choice_text()}"
|
||||||
|
arg_line = Text(f" {flags_choice:<30} ")
|
||||||
|
help_text = arg.help or ""
|
||||||
|
arg_line.append(help_text)
|
||||||
|
self.console.print(arg_line)
|
||||||
|
|
||||||
|
# Epilogue
|
||||||
|
if self.help_epilogue:
|
||||||
|
self.console.print("\n" + self.help_epilogue, style="dim")
|
||||||
|
|
||||||
|
def __eq__(self, other: object) -> bool:
|
||||||
|
if not isinstance(other, CommandArgumentParser):
|
||||||
|
return False
|
||||||
|
|
||||||
|
def sorted_args(parser):
|
||||||
|
return sorted(parser._arguments, key=lambda a: a.dest)
|
||||||
|
|
||||||
|
return sorted_args(self) == sorted_args(other)
|
||||||
|
|
||||||
|
def __hash__(self) -> int:
|
||||||
|
return hash(tuple(sorted(self._arguments, key=lambda a: a.dest)))
|
||||||
|
|
||||||
def __str__(self) -> str:
|
def __str__(self) -> str:
|
||||||
positional = sum(arg.positional for arg in self._arguments)
|
positional = sum(arg.positional for arg in self._arguments)
|
|
@ -0,0 +1,71 @@
|
||||||
|
import inspect
|
||||||
|
from typing import Any, Callable
|
||||||
|
|
||||||
|
from falyx import logger
|
||||||
|
|
||||||
|
|
||||||
|
def infer_args_from_func(
|
||||||
|
func: Callable[[Any], Any],
|
||||||
|
arg_metadata: dict[str, str | dict[str, Any]] | None = None,
|
||||||
|
) -> list[dict[str, Any]]:
|
||||||
|
"""
|
||||||
|
Infer argument definitions from a callable's signature.
|
||||||
|
Returns a list of kwargs suitable for CommandArgumentParser.add_argument.
|
||||||
|
"""
|
||||||
|
arg_metadata = arg_metadata or {}
|
||||||
|
signature = inspect.signature(func)
|
||||||
|
arg_defs = []
|
||||||
|
|
||||||
|
for name, param in signature.parameters.items():
|
||||||
|
raw_metadata = arg_metadata.get(name, {})
|
||||||
|
metadata = (
|
||||||
|
{"help": raw_metadata} if isinstance(raw_metadata, str) else raw_metadata
|
||||||
|
)
|
||||||
|
|
||||||
|
if param.kind not in (
|
||||||
|
inspect.Parameter.POSITIONAL_ONLY,
|
||||||
|
inspect.Parameter.POSITIONAL_OR_KEYWORD,
|
||||||
|
inspect.Parameter.KEYWORD_ONLY,
|
||||||
|
):
|
||||||
|
continue
|
||||||
|
|
||||||
|
arg_type = (
|
||||||
|
param.annotation if param.annotation is not inspect.Parameter.empty else str
|
||||||
|
)
|
||||||
|
default = param.default if param.default is not inspect.Parameter.empty else None
|
||||||
|
is_required = param.default is inspect.Parameter.empty
|
||||||
|
if is_required:
|
||||||
|
flags = [f"{name.replace('_', '-')}"]
|
||||||
|
else:
|
||||||
|
flags = [f"--{name.replace('_', '-')}"]
|
||||||
|
action = "store"
|
||||||
|
nargs: int | str = 1
|
||||||
|
|
||||||
|
if arg_type is bool:
|
||||||
|
if param.default is False:
|
||||||
|
action = "store_true"
|
||||||
|
else:
|
||||||
|
action = "store_false"
|
||||||
|
|
||||||
|
if arg_type is list:
|
||||||
|
action = "append"
|
||||||
|
if is_required:
|
||||||
|
nargs = "+"
|
||||||
|
else:
|
||||||
|
nargs = "*"
|
||||||
|
|
||||||
|
arg_defs.append(
|
||||||
|
{
|
||||||
|
"flags": flags,
|
||||||
|
"dest": name,
|
||||||
|
"type": arg_type,
|
||||||
|
"default": default,
|
||||||
|
"required": is_required,
|
||||||
|
"nargs": nargs,
|
||||||
|
"action": action,
|
||||||
|
"help": metadata.get("help", ""),
|
||||||
|
"choices": metadata.get("choices"),
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
return arg_defs
|
|
@ -0,0 +1,33 @@
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
from falyx import logger
|
||||||
|
from falyx.action.action import Action, ChainedAction, ProcessAction
|
||||||
|
from falyx.parsers.signature import infer_args_from_func
|
||||||
|
|
||||||
|
|
||||||
|
def same_argument_definitions(
|
||||||
|
actions: list[Any],
|
||||||
|
arg_metadata: dict[str, str | dict[str, Any]] | None = None,
|
||||||
|
) -> list[dict[str, Any]] | None:
|
||||||
|
arg_sets = []
|
||||||
|
for action in actions:
|
||||||
|
if isinstance(action, (Action, ProcessAction)):
|
||||||
|
arg_defs = infer_args_from_func(action.action, arg_metadata)
|
||||||
|
elif isinstance(action, ChainedAction):
|
||||||
|
if action.actions:
|
||||||
|
action = action.actions[0]
|
||||||
|
if isinstance(action, Action):
|
||||||
|
arg_defs = infer_args_from_func(action.action, arg_metadata)
|
||||||
|
elif callable(action):
|
||||||
|
arg_defs = infer_args_from_func(action, arg_metadata)
|
||||||
|
elif callable(action):
|
||||||
|
arg_defs = infer_args_from_func(action, arg_metadata)
|
||||||
|
else:
|
||||||
|
logger.debug("Auto args unsupported for action: %s", action)
|
||||||
|
return None
|
||||||
|
arg_sets.append(arg_defs)
|
||||||
|
|
||||||
|
first = arg_sets[0]
|
||||||
|
if all(arg_set == first for arg_set in arg_sets[1:]):
|
||||||
|
return first
|
||||||
|
return None
|
|
@ -1 +1 @@
|
||||||
__version__ = "0.1.28"
|
__version__ = "0.1.29"
|
||||||
|
|
|
@ -1,6 +1,6 @@
|
||||||
[tool.poetry]
|
[tool.poetry]
|
||||||
name = "falyx"
|
name = "falyx"
|
||||||
version = "0.1.28"
|
version = "0.1.29"
|
||||||
description = "Reliable and introspectable async CLI action framework."
|
description = "Reliable and introspectable async CLI action framework."
|
||||||
authors = ["Roland Thomas Jr <roland@rtj.dev>"]
|
authors = ["Roland Thomas Jr <roland@rtj.dev>"]
|
||||||
license = "MIT"
|
license = "MIT"
|
||||||
|
|
|
@ -1,7 +1,7 @@
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from falyx.argparse import ArgumentAction, CommandArgumentParser
|
|
||||||
from falyx.exceptions import CommandArgumentError
|
from falyx.exceptions import CommandArgumentError
|
||||||
|
from falyx.parsers import ArgumentAction, CommandArgumentParser
|
||||||
from falyx.signals import HelpSignal
|
from falyx.signals import HelpSignal
|
||||||
|
|
||||||
|
|
||||||
|
|
Loading…
Reference in New Issue