refactor: Destructured implementation to be more like Arg handling to fix default and helptext behavior.

Author: DanCardin

CHANGELOG.md

@@ -12,6 +12,7 @@
 - fix: Allow negative number arguments and option values.
 - fix: Precalculate implicit deps during class construction rather than traversing the output shape.
 - fix: Thread `output` into `parse_result.instance` resolution in `parse`/`parse_async`/`invoke`/`invoke_async`, so `cappa.Exit` raised from `Arg(parse=...)` callbacks renders an error message instead of silently exiting non-zero.
+- refactor: Destructured implementation to be more like Arg handling to fix default and helptext behavior. 
 
 ## 0.31
 

src/cappa/arg.py

@@ -14,6 +14,7 @@
     Literal,
     Sequence,
     Set,
+    TextIO,
     Union,
     cast,
 )
@@ -24,7 +25,14 @@
 from cappa.completion.completers import complete_choices
 from cappa.completion.types import Completion
 from cappa.default import Default, DefaultFormatter, ValueFrom
-from cappa.parse import Parser, evaluate_parse, parse_literal, parse_value
+from cappa.invoke.types import Resolved
+from cappa.parse import (
+    Parser,
+    evaluate_parse,
+    parse_handler,
+    parse_literal,
+    parse_value,
+)
 from cappa.state import State
 from cappa.type_view import Empty, EmptyType, TypeView
 from cappa.typing import (
@@ -36,7 +44,7 @@
 )
 
 if TYPE_CHECKING:
-    from cappa.destructure import Destructure, destructure
+    from cappa.destructure import Destructure
 
 
 @enum.unique
@@ -418,7 +426,7 @@ def collect(
         default_short: bool = False,
         default_long: bool = False,
         state: State[Any] | None = None,
-    ) -> list[FinalArg[Any]]:
+    ) -> list[FinalArg[Any] | FinalDestructure[Any]]:
         args: list[Arg[Any]] = find_annotations(type_view, cls) or [Arg()]
 
         exclusive = field.name if len(args) > 1 else False
@@ -436,7 +444,7 @@ def collect(
         if field_metadata:
             args = field_metadata
 
-        result: list[FinalArg[Any]] = []
+        result: list[FinalArg[Any] | FinalDestructure[Any]] = []
         for i, arg in enumerate(args, start=1):
             # field_name and default are evaluated outside `normalize` because they can
             # potentially depend on the type's dataclass-like field information, whereas
@@ -458,8 +466,7 @@ def collect(
             )
 
             if normalized_arg.destructure:
-                destructured_args = destructure(normalized_arg, type_view)
-                result.extend(destructured_args)
+                result.extend(normalized_arg.destructure.explode_args())
             else:
                 result.append(normalized_arg)
 
@@ -514,12 +521,9 @@ def normalize(
             show_default if show_default is not None else self.show_default
         )
 
-        destructure = destructure or self.destructure
-        if not destructure:
-            destructure = None
-        if destructure is True:
-            destructure = Destructure()
-
+        destructure = Destructure.collect(
+            field_name, default, destructure or self.destructure, type_view
+        )
         result: FinalArg[Any] = FinalArg(
             # preserved from self
             count=self.count,
@@ -578,7 +582,22 @@ class FinalArg(Arg[T]):
     type_view: TypeView[Any] = dataclasses.field(default_factory=lambda: TypeView(Any))
     parse: Callable[..., Any] = dataclasses.field(default=parse_value)
     show_default: DefaultFormatter = dataclasses.field(default_factory=DefaultFormatter)
-    destructure: Destructure | None = None
+    destructure: FinalDestructure[Any] | None = None
+
+    def map_result(
+        self,
+        prog: str,
+        parsed_args: dict[str, Any],
+        state: State[Any] | None = None,
+        input: TextIO | None = None,
+    ) -> Resolved[T]:
+        field_name = self.field_name
+        if field_name in parsed_args:
+            is_parsed, value = False, parsed_args[field_name]
+        else:
+            is_parsed, value = self.default(state=state, input=input)
+        handler = parse_handler(self.parse, prog, value, self.names_str())
+        return Resolved(handler, args=(value, is_parsed))
 
     def names(self, *, n: int = 0) -> list[str]:
         result = (self.short or []) + (self.long or [])
@@ -878,14 +897,19 @@ def infer_value_name(arg: Arg[Any], field_name: str, num_args: NumArgs) -> str:
 
 
 def explode_negated_bool_args(
-    args: Sequence[FinalArg[Any]],
-) -> Iterable[FinalArg[Any]]:
+    args: Sequence[FinalArg[Any] | FinalDestructure[Any]],
+) -> Iterable[FinalArg[Any] | FinalDestructure[Any]]:
     """Expand `--foo/--no-foo` solo arguments into dual-arguments.
 
     Acts as a transform from `Arg(long='--foo/--no-foo')` to
     `Annotated[Arg(long='--foo', action=ArgAction.store_true), Arg(long='--no-foo', action=ArgAction.store_false)]`
     """
     for arg in args:
+        # Only args can *be* bool negatable args.
+        if not isinstance(arg, FinalArg):
+            yield arg
+            continue
+
         yielded = False
         if isinstance(arg.action, ArgAction) and arg.action.is_bool_action and arg.long:
             long = arg.long
@@ -940,4 +964,4 @@ def infer_has_value(arg: Arg[Any], action: ArgActionType):
     return True
 
 
-from cappa.destructure import Destructure, destructure  # noqa: E402
+from cappa.destructure import Destructure, FinalDestructure  # noqa: E402

src/cappa/command.py

@@ -2,7 +2,6 @@
 
 import contextlib
 import dataclasses
-import inspect
 import sys
 from collections.abc import Callable
 from typing import (
@@ -26,7 +25,6 @@
 from cappa.arg import Arg, FinalArg, Group
 from cappa.class_inspect import fields as get_fields
 from cappa.class_inspect import get_command, get_command_capable_object
-from cappa.default import Default
 from cappa.docstring import ClassHelpText
 from cappa.help import HelpFormattable, HelpFormatter
 from cappa.invoke.types import Resolved
@@ -115,8 +113,8 @@ class Command(Generic[T]):
     """
 
     cmd_cls: type[T]
-    arguments: Sequence[Arg[Any] | Subcommand] = dataclasses.field(
-        default_factory=lambda: []
+    arguments: Sequence[Arg[Any] | Subcommand | FinalDestructure[Any]] = (
+        dataclasses.field(default_factory=lambda: [])
     )
     propagated_arguments: list[FinalArg[Any]] = dataclasses.field(
         default_factory=lambda: []
@@ -192,7 +190,7 @@ def collect(
 
         propagated_arguments = propagated_arguments or []
 
-        arguments: list[FinalArg[Any]] = []
+        arguments: list[FinalArg[Any] | FinalDestructure[Any]] = []
         raw_subcommands: list[tuple[Subcommand, TypeView[Any] | None, str | None]] = []
         if self.arguments:
             param_by_name = {p.name: p for p in function_view.parameters}
@@ -213,6 +211,8 @@ def collect(
                             state=state,
                         )
                     )
+                elif isinstance(arg, FinalDestructure):
+                    arguments.append(arg)
                 else:
                     raw_subcommands.append((arg, None, None))
 
@@ -233,7 +233,7 @@ def collect(
                         )
                     )
                 else:
-                    arg_defs: list[FinalArg[Any]] = Arg.collect(
+                    arg_defs = Arg.collect(
                         field,
                         param_view.type_view,
                         fallback_help=arg_help,
@@ -245,7 +245,7 @@ def collect(
 
         propagating_arguments = [
             *propagated_arguments,
-            *(arg for arg in arguments if arg.propagate),
+            *(arg for arg in arguments if isinstance(arg, FinalArg) and arg.propagate),
         ]
         subcommands = [
             subcommand.normalize(
@@ -258,8 +258,10 @@ def collect(
             for subcommand, type_view, field_name in raw_subcommands
         ]
 
-        check_group_identity(arguments)
-        final_arguments: list[FinalArg[Any] | FinalSubcommand] = [
+        check_group_identity([a for a in arguments if isinstance(a, FinalArg)])
+        final_arguments: list[
+            FinalArg[Any] | FinalSubcommand | FinalDestructure[Any]
+        ] = [
             *arguments,
             *subcommands,
         ]
@@ -290,8 +292,8 @@ class FinalCommand(Command[T]):
     Produced exclusively by :meth:`Command.collect`.
     """
 
-    arguments: Sequence[FinalArg[Any] | FinalSubcommand] = dataclasses.field(  # pyright: ignore
-        default_factory=list
+    arguments: Sequence[FinalArg[Any] | FinalSubcommand | FinalDestructure[Any]] = (  # pyright: ignore
+        dataclasses.field(default_factory=list)
     )
     propagated_arguments: list[FinalArg[Any]] = dataclasses.field(  # pyright: ignore
         default_factory=list
@@ -310,10 +312,17 @@ def value_arguments(self) -> Iterable[FinalArg[Any]]:
             if isinstance(arg, FinalArg) and arg.has_value:
                 yield arg
 
+    @property
+    def destructured_arguments(self) -> Iterable[FinalDestructure[Any]]:
+        for arg in self.arguments:
+            if isinstance(arg, FinalDestructure):
+                yield arg
+
     @property
     def all_arguments(self) -> Iterable[FinalArg[Any] | FinalSubcommand]:
         for arg in self.arguments:
-            yield arg
+            if not isinstance(arg, FinalDestructure):
+                yield arg
         for arg in self.propagated_arguments:
             yield arg
 
@@ -329,10 +338,7 @@ def positional_arguments(
     ) -> Iterable[FinalArg[Any] | FinalSubcommand]:
         for arg in self.arguments:
             if (
-                isinstance(arg, FinalArg)
-                and not arg.short
-                and not arg.long
-                and not arg.destructure
+                isinstance(arg, FinalArg) and not arg.short and not arg.long
             ) or isinstance(arg, FinalSubcommand):
                 yield arg
 
@@ -379,18 +385,18 @@ def map_result(
 
         kwargs: dict[str, Any] = {}
         for arg in self.value_arguments:
-            field_name = arg.field_name
-
-            if field_name in parsed_args:
-                is_parsed, value = (False, parsed_args[field_name])
-            else:
-                assert isinstance(arg.default, Default), arg
-                is_parsed, value = arg.default(state=state, input=input)
-
-            handler = parse_handler(arg, prog, value)
-            kwargs[field_name] = Resolved(handler, args=(value, is_parsed))
+            kwargs[arg.field_name] = arg.map_result(
+                prog, parsed_args, state=state, input=input
+            )
 
         subcommand_deps: dict[Hashable, Any] = {}
+        for destructure in self.destructured_arguments:
+            fd_parsed = parsed_args.get(destructure.field_name, {})
+            fd_resolved = destructure.map_result(
+                prog, fd_parsed, output, state=state, input=input
+            )
+            kwargs[destructure.field_name] = fd_resolved
+
         subcommand = self.subcommand
         if subcommand:
             field_name = subcommand.field_name
@@ -464,62 +470,6 @@ def check_group_identity(args: list[FinalArg[Any]]):
         group_identity[arg.group.id] = arg.group
 
 
-@contextlib.contextmanager
-def parse_value(
-    arg: FinalArg[T], prog: str, value: Any, is_parsed: bool
-) -> Generator[T, None, None]:
-    if not is_parsed:
-        try:
-            value = arg.parse(value)
-        except Exception as e:
-            exception_reason = str(e)
-            raise Exit(
-                f"Invalid value for '{arg.names_str()}': {exception_reason}",
-                code=2,
-                prog=prog,
-            )
-
-    yield value
-
-
-def parse_handler(
-    arg: FinalArg[T], prog: str, value: Any
-) -> Callable[[Any, bool], Any]:
-    is_async_value = inspect.iscoroutine(value)
-    is_async_parse = inspect.iscoroutinefunction(
-        arg.parse
-    ) or inspect.isasyncgenfunction(arg.parse)
-
-    if is_async_value or is_async_parse:
-
-        async def async_parse(raw_value: Any, is_parsed: bool) -> T:
-            # Await the value if it's a coroutine
-            if inspect.iscoroutine(raw_value):
-                raw_value = await raw_value
-
-            with parse_value(arg, prog, raw_value, is_parsed) as parsed:
-                # Check if the parse result is a coroutine and await it
-                if inspect.iscoroutine(parsed):
-                    try:
-                        return await parsed
-                    except Exception as e:
-                        exception_reason = str(e)
-                        raise Exit(
-                            f"Invalid value for '{arg.names_str()}': {exception_reason}",
-                            code=2,
-                            prog=prog,
-                        )
-                return parsed
-
-        return async_parse
-
-    def sync_parse(raw_value: Any, is_parsed: bool) -> T:
-        with parse_value(arg, prog, raw_value, is_parsed) as parsed:
-            return parsed
-
-    return sync_parse
-
-
 @contextlib.contextmanager
 def graceful_exit(
     command: FinalCommand[T], prog: str, output: Output
@@ -544,3 +494,6 @@ def graceful_exit(
             raise
 
         raise
+
+
+from cappa.destructure import FinalDestructure  # noqa: E402

src/cappa/default.py

@@ -115,7 +115,7 @@ def __or__(self, other: Any) -> Default:
 
     def __call__(
         self, state: State[Any] | None = None, input: TextIO | None = None
-    ) -> tuple[bool, Any | None]:
+    ) -> tuple[bool, Any]:
         """Evaluate the default retrieval sequence, returning the first non-Empty value."""
         for default in self.sequence:
             if isinstance(default, ValueFrom):