Centralize CAP creation in Command, Add better default type coercion
This commit is contained in:
parent
21af003bc7
commit
1c97857cb8
|
@ -128,7 +128,7 @@ class Command(BaseModel):
|
|||
tags: list[str] = Field(default_factory=list)
|
||||
logging_hooks: bool = False
|
||||
options_manager: OptionsManager = Field(default_factory=OptionsManager)
|
||||
arg_parser: CommandArgumentParser = Field(default_factory=CommandArgumentParser)
|
||||
arg_parser: CommandArgumentParser | None = None
|
||||
arguments: list[dict[str, Any]] = Field(default_factory=list)
|
||||
argument_config: Callable[[CommandArgumentParser], None] | None = None
|
||||
custom_parser: ArgParserProtocol | None = None
|
||||
|
@ -167,6 +167,12 @@ class Command(BaseModel):
|
|||
raw_args,
|
||||
)
|
||||
return ((), {})
|
||||
if not isinstance(self.arg_parser, CommandArgumentParser):
|
||||
logger.warning(
|
||||
"[Command:%s] No argument parser configured, using default parsing.",
|
||||
self.key,
|
||||
)
|
||||
return ((), {})
|
||||
return await self.arg_parser.parse_args_split(
|
||||
raw_args, from_validate=from_validate
|
||||
)
|
||||
|
@ -183,7 +189,9 @@ class Command(BaseModel):
|
|||
def get_argument_definitions(self) -> list[dict[str, Any]]:
|
||||
if self.arguments:
|
||||
return self.arguments
|
||||
elif callable(self.argument_config):
|
||||
elif callable(self.argument_config) and isinstance(
|
||||
self.arg_parser, CommandArgumentParser
|
||||
):
|
||||
self.argument_config(self.arg_parser)
|
||||
elif self.auto_args:
|
||||
if isinstance(self.action, BaseAction):
|
||||
|
@ -219,8 +227,17 @@ class Command(BaseModel):
|
|||
if self.logging_hooks and isinstance(self.action, BaseAction):
|
||||
register_debug_hooks(self.action.hooks)
|
||||
|
||||
for arg_def in self.get_argument_definitions():
|
||||
self.arg_parser.add_argument(*arg_def.pop("flags"), **arg_def)
|
||||
if self.arg_parser is None:
|
||||
self.arg_parser = CommandArgumentParser(
|
||||
command_key=self.key,
|
||||
command_description=self.description,
|
||||
command_style=self.style,
|
||||
help_text=self.help_text,
|
||||
help_epilogue=self.help_epilogue,
|
||||
aliases=self.aliases,
|
||||
)
|
||||
for arg_def in self.get_argument_definitions():
|
||||
self.arg_parser.add_argument(*arg_def.pop("flags"), **arg_def)
|
||||
|
||||
def _inject_options_manager(self) -> None:
|
||||
"""Inject the options manager into the action if applicable."""
|
||||
|
|
|
@ -118,14 +118,6 @@ def convert_commands(raw_commands: list[dict[str, Any]]) -> list[Command]:
|
|||
commands = []
|
||||
for entry in raw_commands:
|
||||
raw_command = RawCommand(**entry)
|
||||
parser = CommandArgumentParser(
|
||||
command_key=raw_command.key,
|
||||
command_description=raw_command.description,
|
||||
command_style=raw_command.style,
|
||||
help_text=raw_command.help_text,
|
||||
help_epilogue=raw_command.help_epilogue,
|
||||
aliases=raw_command.aliases,
|
||||
)
|
||||
commands.append(
|
||||
Command.model_validate(
|
||||
{
|
||||
|
@ -133,7 +125,6 @@ def convert_commands(raw_commands: list[dict[str, Any]]) -> list[Command]:
|
|||
"action": wrap_if_needed(
|
||||
import_action(raw_command.action), name=raw_command.description
|
||||
),
|
||||
"arg_parser": parser,
|
||||
}
|
||||
)
|
||||
)
|
||||
|
|
|
@ -359,7 +359,6 @@ class Falyx:
|
|||
action=Action("Help", self._show_help),
|
||||
style=OneColors.LIGHT_YELLOW,
|
||||
arg_parser=parser,
|
||||
auto_args=False,
|
||||
)
|
||||
|
||||
def _get_completer(self) -> WordCompleter:
|
||||
|
@ -655,15 +654,6 @@ class Falyx:
|
|||
"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(
|
||||
key=key,
|
||||
|
|
|
@ -12,6 +12,7 @@ from rich.text import Text
|
|||
|
||||
from falyx.action.base import BaseAction
|
||||
from falyx.exceptions import CommandArgumentError
|
||||
from falyx.parsers.utils import coerce_value
|
||||
from falyx.signals import HelpSignal
|
||||
|
||||
|
||||
|
@ -290,7 +291,7 @@ class CommandArgumentParser:
|
|||
for choice in choices:
|
||||
if not isinstance(choice, expected_type):
|
||||
try:
|
||||
expected_type(choice)
|
||||
coerce_value(choice, expected_type)
|
||||
except Exception:
|
||||
raise CommandArgumentError(
|
||||
f"Invalid choice {choice!r}: not coercible to {expected_type.__name__}"
|
||||
|
@ -303,7 +304,7 @@ class CommandArgumentParser:
|
|||
"""Validate the default value type."""
|
||||
if default is not None and not isinstance(default, expected_type):
|
||||
try:
|
||||
expected_type(default)
|
||||
coerce_value(default, expected_type)
|
||||
except Exception:
|
||||
raise CommandArgumentError(
|
||||
f"Default value {default!r} for '{dest}' cannot be coerced to {expected_type.__name__}"
|
||||
|
@ -316,7 +317,7 @@ class CommandArgumentParser:
|
|||
for item in default:
|
||||
if not isinstance(item, expected_type):
|
||||
try:
|
||||
expected_type(item)
|
||||
coerce_value(item, expected_type)
|
||||
except Exception:
|
||||
raise CommandArgumentError(
|
||||
f"Default list value {default!r} for '{dest}' cannot be coerced to {expected_type.__name__}"
|
||||
|
@ -595,7 +596,7 @@ class CommandArgumentParser:
|
|||
i += new_i
|
||||
|
||||
try:
|
||||
typed = [spec.type(v) for v in values]
|
||||
typed = [coerce_value(value, spec.type) for value in values]
|
||||
except Exception:
|
||||
raise CommandArgumentError(
|
||||
f"Invalid value for '{spec.dest}': expected {spec.type.__name__}"
|
||||
|
@ -680,7 +681,9 @@ class CommandArgumentParser:
|
|||
), "resolver should be an instance of BaseAction"
|
||||
values, new_i = self._consume_nargs(args, i + 1, spec)
|
||||
try:
|
||||
typed_values = [spec.type(value) for value in values]
|
||||
typed_values = [
|
||||
coerce_value(value, spec.type) for value in values
|
||||
]
|
||||
except ValueError:
|
||||
raise CommandArgumentError(
|
||||
f"Invalid value for '{spec.dest}': expected {spec.type.__name__}"
|
||||
|
@ -709,7 +712,9 @@ class CommandArgumentParser:
|
|||
assert result.get(spec.dest) is not None, "dest should not be None"
|
||||
values, new_i = self._consume_nargs(args, i + 1, spec)
|
||||
try:
|
||||
typed_values = [spec.type(value) for value in values]
|
||||
typed_values = [
|
||||
coerce_value(value, spec.type) for value in values
|
||||
]
|
||||
except ValueError:
|
||||
raise CommandArgumentError(
|
||||
f"Invalid value for '{spec.dest}': expected {spec.type.__name__}"
|
||||
|
@ -724,7 +729,9 @@ class CommandArgumentParser:
|
|||
assert result.get(spec.dest) is not None, "dest should not be None"
|
||||
values, new_i = self._consume_nargs(args, i + 1, spec)
|
||||
try:
|
||||
typed_values = [spec.type(value) for value in values]
|
||||
typed_values = [
|
||||
coerce_value(value, spec.type) for value in values
|
||||
]
|
||||
except ValueError:
|
||||
raise CommandArgumentError(
|
||||
f"Invalid value for '{spec.dest}': expected {spec.type.__name__}"
|
||||
|
@ -735,7 +742,9 @@ class CommandArgumentParser:
|
|||
else:
|
||||
values, new_i = self._consume_nargs(args, i + 1, spec)
|
||||
try:
|
||||
typed_values = [spec.type(v) for v in values]
|
||||
typed_values = [
|
||||
coerce_value(value, spec.type) for value in values
|
||||
]
|
||||
except ValueError:
|
||||
raise CommandArgumentError(
|
||||
f"Invalid value for '{spec.dest}': expected {spec.type.__name__}"
|
||||
|
|
|
@ -1,10 +1,79 @@
|
|||
from typing import Any
|
||||
import types
|
||||
from datetime import datetime
|
||||
from enum import EnumMeta
|
||||
from typing import Any, Literal, Union, get_args, get_origin
|
||||
|
||||
from dateutil import parser as date_parser
|
||||
|
||||
from falyx.action.base import BaseAction
|
||||
from falyx.logger import logger
|
||||
from falyx.parsers.signature import infer_args_from_func
|
||||
|
||||
|
||||
def coerce_bool(value: str) -> bool:
|
||||
if isinstance(value, bool):
|
||||
return value
|
||||
value = value.strip().lower()
|
||||
if value in {"true", "1", "yes", "on"}:
|
||||
return True
|
||||
elif value in {"false", "0", "no", "off"}:
|
||||
return False
|
||||
return bool(value)
|
||||
|
||||
|
||||
def coerce_enum(value: Any, enum_type: EnumMeta) -> Any:
|
||||
if isinstance(value, enum_type):
|
||||
return value
|
||||
|
||||
if isinstance(value, str):
|
||||
try:
|
||||
return enum_type[value]
|
||||
except KeyError:
|
||||
pass
|
||||
|
||||
base_type = type(next(iter(enum_type)).value)
|
||||
print(base_type)
|
||||
try:
|
||||
coerced_value = base_type(value)
|
||||
return enum_type(coerced_value)
|
||||
except (ValueError, TypeError):
|
||||
raise ValueError(f"Value '{value}' could not be coerced to enum type {enum_type}")
|
||||
|
||||
|
||||
def coerce_value(value: str, target_type: type) -> Any:
|
||||
origin = get_origin(target_type)
|
||||
args = get_args(target_type)
|
||||
|
||||
if origin is Literal:
|
||||
if value not in args:
|
||||
raise ValueError(
|
||||
f"Value '{value}' is not a valid literal for type {target_type}"
|
||||
)
|
||||
return value
|
||||
|
||||
if isinstance(target_type, types.UnionType) or get_origin(target_type) is Union:
|
||||
for arg in args:
|
||||
try:
|
||||
return coerce_value(value, arg)
|
||||
except Exception:
|
||||
continue
|
||||
raise ValueError(f"Value '{value}' could not be coerced to any of {args!r}")
|
||||
|
||||
if isinstance(target_type, EnumMeta):
|
||||
return coerce_enum(value, target_type)
|
||||
|
||||
if target_type is bool:
|
||||
return coerce_bool(value)
|
||||
|
||||
if target_type is datetime:
|
||||
try:
|
||||
return date_parser.parse(value)
|
||||
except ValueError as e:
|
||||
raise ValueError(f"Value '{value}' could not be parsed as a datetime") from e
|
||||
|
||||
return target_type(value)
|
||||
|
||||
|
||||
def same_argument_definitions(
|
||||
actions: list[Any],
|
||||
arg_metadata: dict[str, str | dict[str, Any]] | None = None,
|
||||
|
|
|
@ -1 +1 @@
|
|||
__version__ = "0.1.44"
|
||||
__version__ = "0.1.45"
|
||||
|
|
|
@ -1,6 +1,6 @@
|
|||
[tool.poetry]
|
||||
name = "falyx"
|
||||
version = "0.1.44"
|
||||
version = "0.1.45"
|
||||
description = "Reliable and introspectable async CLI action framework."
|
||||
authors = ["Roland Thomas Jr <roland@rtj.dev>"]
|
||||
license = "MIT"
|
||||
|
@ -16,6 +16,7 @@ python-json-logger = "^3.3.0"
|
|||
toml = "^0.10"
|
||||
pyyaml = "^6.0"
|
||||
aiohttp = "^3.11"
|
||||
python-dateutil = "^2.8"
|
||||
|
||||
[tool.poetry.group.dev.dependencies]
|
||||
pytest = "^8.3.5"
|
||||
|
|
|
@ -0,0 +1,153 @@
|
|||
from datetime import datetime
|
||||
from enum import Enum
|
||||
from pathlib import Path
|
||||
from typing import Literal
|
||||
|
||||
import pytest
|
||||
|
||||
from falyx.parsers.utils import coerce_value
|
||||
|
||||
|
||||
# --- Tests ---
|
||||
@pytest.mark.parametrize(
|
||||
"value, target_type, expected",
|
||||
[
|
||||
("42", int, 42),
|
||||
("3.14", float, 3.14),
|
||||
("True", bool, True),
|
||||
("hello", str, "hello"),
|
||||
("", str, ""),
|
||||
("False", bool, False),
|
||||
],
|
||||
)
|
||||
def test_coerce_value_basic(value, target_type, expected):
|
||||
assert coerce_value(value, target_type) == expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"value, target_type, expected",
|
||||
[
|
||||
("42", int | float, 42),
|
||||
("3.14", int | float, 3.14),
|
||||
("hello", str | int, "hello"),
|
||||
("1", bool | str, True),
|
||||
],
|
||||
)
|
||||
def test_coerce_value_union_success(value, target_type, expected):
|
||||
assert coerce_value(value, target_type) == expected
|
||||
|
||||
|
||||
def test_coerce_value_union_failure():
|
||||
with pytest.raises(ValueError) as excinfo:
|
||||
coerce_value("abc", int | float)
|
||||
assert "could not be coerced" in str(excinfo.value)
|
||||
|
||||
|
||||
def test_coerce_value_typing_union_equivalent():
|
||||
from typing import Union
|
||||
|
||||
assert coerce_value("123", Union[int, str]) == 123
|
||||
assert coerce_value("abc", Union[int, str]) == "abc"
|
||||
|
||||
|
||||
def test_coerce_value_edge_cases():
|
||||
# int -> raises
|
||||
with pytest.raises(ValueError):
|
||||
coerce_value("not-an-int", int | float)
|
||||
|
||||
# empty string with str fallback
|
||||
assert coerce_value("", int | str) == ""
|
||||
|
||||
# bool conversion
|
||||
assert coerce_value("False", bool | str) is False
|
||||
|
||||
|
||||
def test_coerce_value_enum():
|
||||
class Color(Enum):
|
||||
RED = "red"
|
||||
GREEN = "green"
|
||||
BLUE = "blue"
|
||||
|
||||
assert coerce_value("red", Color) == Color.RED
|
||||
assert coerce_value("green", Color) == Color.GREEN
|
||||
assert coerce_value("blue", Color) == Color.BLUE
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
coerce_value("yellow", Color) # Not a valid enum value
|
||||
|
||||
|
||||
def test_coerce_value_int_enum():
|
||||
class Status(Enum):
|
||||
SUCCESS = 0
|
||||
FAILURE = 1
|
||||
PENDING = 2
|
||||
|
||||
assert coerce_value("0", Status) == Status.SUCCESS
|
||||
assert coerce_value(1, Status) == Status.FAILURE
|
||||
assert coerce_value("PENDING", Status) == Status.PENDING
|
||||
assert coerce_value(Status.SUCCESS, Status) == Status.SUCCESS
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
coerce_value("3", Status)
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
coerce_value(3, Status)
|
||||
|
||||
|
||||
class Mode(Enum):
|
||||
DEV = "dev"
|
||||
PROD = "prod"
|
||||
|
||||
|
||||
def test_literal_coercion():
|
||||
assert coerce_value("dev", Literal["dev", "prod"]) == "dev"
|
||||
try:
|
||||
coerce_value("staging", Literal["dev", "prod"])
|
||||
assert False
|
||||
except ValueError:
|
||||
assert True
|
||||
|
||||
|
||||
def test_enum_coercion():
|
||||
assert coerce_value("dev", Mode) == Mode.DEV
|
||||
assert coerce_value("DEV", Mode) == Mode.DEV
|
||||
try:
|
||||
coerce_value("staging", Mode)
|
||||
assert False
|
||||
except ValueError:
|
||||
assert True
|
||||
|
||||
|
||||
def test_union_coercion():
|
||||
assert coerce_value("123", int | str) == 123
|
||||
assert coerce_value("abc", int | str) == "abc"
|
||||
assert coerce_value("False", bool | str) is False
|
||||
|
||||
|
||||
def test_path_coercion():
|
||||
result = coerce_value("/tmp/test.txt", Path)
|
||||
assert isinstance(result, Path)
|
||||
assert str(result) == "/tmp/test.txt"
|
||||
|
||||
|
||||
def test_datetime_coercion():
|
||||
result = coerce_value("2023-10-01T13:00:00", datetime)
|
||||
assert isinstance(result, datetime)
|
||||
assert result.year == 2023 and result.month == 10
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
coerce_value("not-a-date", datetime)
|
||||
|
||||
|
||||
def test_bool_coercion():
|
||||
assert coerce_value("true", bool) is True
|
||||
assert coerce_value("False", bool) is False
|
||||
assert coerce_value("0", bool) is False
|
||||
assert coerce_value("", bool) is False
|
||||
assert coerce_value("1", bool) is True
|
||||
assert coerce_value("yes", bool) is True
|
||||
assert coerce_value("no", bool) is False
|
||||
assert coerce_value("on", bool) is True
|
||||
assert coerce_value("off", bool) is False
|
||||
assert coerce_value(True, bool) is True
|
||||
assert coerce_value(False, bool) is False
|
Loading…
Reference in New Issue