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

Expose all argument parsers to enable custom logic

Browse Source
This commit is contained in:
Roland Thomas Jr
2025-04-17 21:49:18 -04:00
parent f14b54743f
commit 74555c4484
2 changed files with 72 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
falyx/falyx.py
View File

@@ -15,7 +15,7 @@ user input and create an interactive experience.
import asyncio
import logging
import sys
from argparse import ArgumentParser, Namespace
from argparse import Namespace
from difflib import get_close_matches
from functools import cached_property
from typing import Any, Callable
@@ -39,6 +39,7 @@ from falyx.exceptions import (CommandAlreadyExistsError, FalyxError,
InvalidActionError, NotAFalyxError)
from falyx.execution_registry import ExecutionRegistry as er
from falyx.hook_manager import Hook, HookManager, HookType
from falyx.parsers import FalyxParsers, get_arg_parsers
from falyx.retry import RetryPolicy
from falyx.themes.colors import OneColors, get_nord_theme
from falyx.utils import CaseInsensitiveDict, async_confirm, chunks, logger
@@ -720,41 +721,6 @@ class Falyx:
selected_command.retry_policy.backoff = self.cli_args.retry_backoff
selected_command.update_retry_policy(selected_command.retry_policy)
def get_arg_parser(self) -> ArgumentParser:
"""Returns the argument parser for the CLI."""
parser = ArgumentParser(prog="falyx", description="Falyx CLI - Run structured async command workflows.")
parser.add_argument("-v", "--verbose", action="store_true", help="Enable debug logging for Falyx.")
parser.add_argument("--debug-hooks", action="store_true", help="Enable default lifecycle debug logging")
parser.add_argument("--version", action="store_true", help="Show Falyx version")
subparsers = parser.add_subparsers(dest="command")
run_parser = subparsers.add_parser("run", help="Run a specific command")
run_parser.add_argument("name", help="Key, alias, or description of the command")
run_parser.add_argument("--retries", type=int, help="Number of retries on failure", default=0)
run_parser.add_argument("--retry-delay", type=float, help="Initial delay between retries in (seconds)", default=0)
run_parser.add_argument("--retry-backoff", type=float, help="Backoff factor for retries", default=0)
run_group = run_parser.add_mutually_exclusive_group(required=False)
run_group.add_argument("-c", "--confirm", dest="force_confirm", action="store_true", help="Force confirmation prompts")
run_group.add_argument("-s", "--skip-confirm", dest="skip_confirm", action="store_true", help="Skip confirmation prompts")
run_all_parser = subparsers.add_parser("run-all", help="Run all commands with a given tag")
run_all_parser.add_argument("-t", "--tag", required=True, help="Tag to match")
run_all_parser.add_argument("--retries", type=int, help="Number of retries on failure", default=0)
run_all_parser.add_argument("--retry-delay", type=float, help="Initial delay between retries in (seconds)", default=0)
run_all_parser.add_argument("--retry-backoff", type=float, help="Backoff factor for retries", default=0)
run_all_group = run_all_parser.add_mutually_exclusive_group(required=False)
run_all_group.add_argument("-c", "--confirm", dest="force_confirm", action="store_true", help="Force confirmation prompts")
run_all_group.add_argument("-s", "--skip-confirm", dest="skip_confirm", action="store_true", help="Skip confirmation prompts")
preview_parser = subparsers.add_parser("preview", help="Preview a command without running it")
preview_parser.add_argument("name", help="Key, alias, or description of the command")
subparsers.add_parser("list", help="List all available commands with tags")
subparsers.add_parser("version", help="Show the Falyx version")
return parser
def print_message(self, message: str | Markdown | dict[str, Any]) -> None:
"""Prints a message to the console."""
if isinstance(message, (str, Markdown)):
@@ -789,10 +755,10 @@ class Falyx:
if self.exit_message:
self.print_message(self.exit_message)
async def run(self, parser: ArgumentParser | None = None) -> None:
async def run(self, parsers: FalyxParsers | None = None) -> None:
"""Run Falyx CLI with structured subcommands."""
parser = parser or self.get_arg_parser()
self.cli_args = parser.parse_args()
parsers = parsers or get_arg_parsers()
self.cli_args = parsers.root.parse_args()
if self.cli_args.verbose:
logging.getLogger("falyx").setLevel(logging.DEBUG)

67
falyx/parsers.py Normal file
View File

@@ -0,0 +1,67 @@
"""parsers.py
This module contains the argument parsers used for the Falyx CLI.
"""
from dataclasses import asdict, dataclass
from argparse import ArgumentParser
@dataclass
class FalyxParsers:
"""Defines the argument parsers for the Falyx CLI."""
root: ArgumentParser
run: ArgumentParser
run_all: ArgumentParser
preview: ArgumentParser
list: ArgumentParser
version: ArgumentParser
def as_dict(self) -> dict[str, ArgumentParser]:
"""Convert the FalyxParsers instance to a dictionary."""
return asdict(self)
def get_parser(self, name: str) -> ArgumentParser | None:
"""Get the parser by name."""
return self.as_dict().get(name)
def get_arg_parsers() -> FalyxParsers:
"""Returns the argument parser for the CLI."""
parser = ArgumentParser(prog="falyx", description="Falyx CLI - Run structured async command workflows.")
parser.add_argument("-v", "--verbose", action="store_true", help="Enable debug logging for Falyx.")
parser.add_argument("--debug-hooks", action="store_true", help="Enable default lifecycle debug logging")
parser.add_argument("--version", action="store_true", help="Show Falyx version")
subparsers = parser.add_subparsers(dest="command")
run_parser = subparsers.add_parser("run", help="Run a specific command")
run_parser.add_argument("name", help="Key, alias, or description of the command")
run_parser.add_argument("--retries", type=int, help="Number of retries on failure", default=0)
run_parser.add_argument("--retry-delay", type=float, help="Initial delay between retries in (seconds)", default=0)
run_parser.add_argument("--retry-backoff", type=float, help="Backoff factor for retries", default=0)
run_group = run_parser.add_mutually_exclusive_group(required=False)
run_group.add_argument("-c", "--confirm", dest="force_confirm", action="store_true", help="Force confirmation prompts")
run_group.add_argument("-s", "--skip-confirm", dest="skip_confirm", action="store_true", help="Skip confirmation prompts")
run_all_parser = subparsers.add_parser("run-all", help="Run all commands with a given tag")
run_all_parser.add_argument("-t", "--tag", required=True, help="Tag to match")
run_all_parser.add_argument("--retries", type=int, help="Number of retries on failure", default=0)
run_all_parser.add_argument("--retry-delay", type=float, help="Initial delay between retries in (seconds)", default=0)
run_all_parser.add_argument("--retry-backoff", type=float, help="Backoff factor for retries", default=0)
run_all_group = run_all_parser.add_mutually_exclusive_group(required=False)
run_all_group.add_argument("-c", "--confirm", dest="force_confirm", action="store_true", help="Force confirmation prompts")
run_all_group.add_argument("-s", "--skip-confirm", dest="skip_confirm", action="store_true", help="Skip confirmation prompts")
preview_parser = subparsers.add_parser("preview", help="Preview a command without running it")
preview_parser.add_argument("name", help="Key, alias, or description of the command")
list_parser = subparsers.add_parser("list", help="List all available commands with tags")
version_parser = subparsers.add_parser("version", help="Show the Falyx version")
return FalyxParsers(
root=parser,
run=run_parser,
run_all=run_all_parser,
preview=preview_parser,
list=list_parser,
version=version_parser,
)